跳到主要内容

QuickCheck 教程 Part 3

· 阅读需 22 分钟

本篇是 QuickCheck 系列的最后一部分,关注的是性质失败之后会发生什么: 框架如何把一个随机失败样本压缩成更小的反例, 以及我们如何判断这个反例是否真的说明了问题。

在工程实践里,随机测试能否真正落地,很大程度上取决于失败质量。 一个又大又乱的失败输入,也许能说明程序有 bug,却未必能降低定位成本; 而一个足够小、足够稳定的反例,往往就已经是一条直接可用的调试线索。

当然,缩减并不是第三篇唯一关心的主题。 受限输入一旦变复杂,手写 generator 与 shrinker 也会迅速变得脆弱。 这时我们就需要更系统的方法,例如小规模穷举、functional enumeration, 以及更进一步的归纳关系方法。

反例与缩减​

在 QuickCheck 中,失败从来不是终点,而是分析的起点。 一个性质失败以后,我们真正关心的是:它为什么失败,最小失败条件是什么。 缩减策略的作用,就是在保持失败事实不变的前提下,把样本压缩到更接近缺陷本质的形式。

从执行流程上看,QuickCheck 会先生成样本、找到失败,再沿着 shrink 树继续搜索更小的失败样本。 这个过程并不会改变性质本身,却会显著改变我们理解 bug 的速度。 很多时候,一个「失败但很大」的反例几乎没有帮助,而一个「更小但语义完整」的反例, 已经足够直接指向实现错误。 因此 shrink 并不是 QuickCheck 的边角功能,它和生成器一样,决定了整个测试体系的可用性。 接下来我们先从 trait Shrink 开始, 了解框架如何抽象这一过程。

默认 Shrink​

Shrink 是 QuickCheck 中负责「把一个值变简单」的 trait。 从签名上看,它的核心方法 shrink 接受一个值,返回一个迭代器,产出一系列「更小的」候选值。 这里使用迭代器是因为我们期望一种「惰性搜索」的语义, 很可能在某个候选值上就能找到一个更小的失败样本,而不需要把所有候选值都枚举出来。

///|
pub trait Shrink {
  shrink(Self) -> Iter[Self]
}

对于大多数基础类型与常见容器,框架已经提供了默认实例,因此只要我们使用 @qc.quick_check_fn 或 @qc.Arrow 这类入口,就会自动启用对应的 shrink 逻辑。 整数会倾向于向 0 靠拢,布尔值会朝 false 收缩,数组与列表则会同时尝试 「删元素」与「缩元素」,元组会按分量继续递归缩减。

先看一个最简单的例子。对整数而言,默认 shrink 的方向并不是盲目枚举所有更小的数, 而是用较少的候选点快速逼近「更简单」的值。

///|
test "shrink int sample" {
  json_inspect(@qc.Shrink::shrink(100), content=[99, 97, 94, 88, 75, 50, 0])
}

这表明 shrink 并不是在做随意扰动,而是在朝着更简单的值组织搜索。 这种思路在容器类型上会更明显:数组会先尝试删除元素, 如果删除还不足以保留失败,再继续缩减仍然相关的位置。

我们回到一个更贴近真实 bug 的例子。假设我们写了一个有缺陷的删除函数, 它只删除数组中第一次出现的元素,而不是删掉全部出现位置:

///|
fn remove_first_only(arr : Array[Int], x : Int) -> Array[Int] {
  guard arr.search(x) is Some(i) else { arr }
  arr.remove(i) |> ignore
  arr
}

///|
fn prop_remove_all(iarr : (Int, Array[Int])) -> Bool {
  let (x, arr) = iarr
  !remove_first_only(arr, x).contains(x)
}

///|
test "default shrink for tuple and array" {
  let x = @qc.quick_check_fn_silence(prop_remove_all)
  inspect(
    x,
    content=(
      #|*** [8/0/100] Failed! Falsified.
      #|(0, [0, 0, -1])
    ),
  )
}

这里 quick_check_fn 会自动组合元组、数组与整数的默认 shrink 逻辑。 直观地说,Int 会不断尝试向 0 靠拢,数组则会优先删去无关元素, 最后再缩减仍然必要的位置。最终反例是 (0, [0, 0, -1]), 它已经很小,并且直接指出了问题核心: remove_first_only 没有正确处理数组里多个 0 的情况。

默认 shrink 在大多数基础场景已经足够好,但它也不是无限制运行的。 @qc.quick_check 与 @qc.quick_check_fn 都提供了 max_shrink / max_shrinks 用于限制缩减预算,这在输入结构很大、缩减树很宽时尤其重要。 换句话说,缩减本身也是一种搜索,我们同样需要在「更小的反例」与「更快的反馈」之间做工程上的权衡。

自定义 Shrink​

默认 shrink 的优势在于通用,但通用的代价是它并不了解业务不变量。 一旦输入带有额外语义,例如「必须是偶数」、「必须非负」和「必须保持有序」, 默认 shrink 就可能把样本缩到一个类型上合法、但业务上并不成立的值。 这时我们就需要显式接管 shrink 过程。

QuickCheck 提供了两个直接相关的接口:

fn[T : Testable, A : Show] @qc.forall_shrink(
  gen : @qc.Gen[A],
  shrinker : (A) -> Iter[A],
  f : (A) -> T,
) -> @qc.Property

fn[P : Testable, T] @qc.shrinking(
  shrinker : (T) -> Iter[T],
  x0 : T,
  pf : (T) -> P,
) -> @qc.Property

forall_shrink 的角色是「给定生成器之后,再显式指定如何缩减生成值」; 而 shrinking 则更底层一些,它不依赖随机生成,而是从一个给定值出发, 直接沿着 shrinker 生成的候选值向下搜索。 前者适合日常性质测试,后者则很适合我们单独调试 shrink 逻辑本身。

下面看一个简单但典型的例子。假设我们的输入域被限定为「非负偶数」, 那么 shrink 时如果跑出奇数,其实就已经偏离了测试对象的语义边界。 这时我们可以在默认 Int shrink 的基础上再加一层过滤:

///|
fn shrink_even_nat(x : Int) -> Iter[Int] {
  @qc.Shrink::shrink(x).filter(fn(y) { y >= 0 && y % 2 == 0 })
}

///|
test "forall_shrink keeps even invariant" {
  let gen = @qc.int_range(0, 100).fmap(fn(x) { x * 2 })
  let prop = @qc.forall_shrink(gen, shrink_even_nat, fn(x) { x < 20 })
  @qc.quick_check(prop, expect=Fail)
}

这里生成器只会产生偶数,而自定义 shrinker 也保证缩减过程中仍然停留在这个域里。 如果我们直接使用默认 Int shrink,框架虽然同样能找到更小的反例, 但中间会穿过大量奇数值,它们在类型上合法,却并不属于我们真正关心的输入语义。 这类「缩得更小但缩偏了」的反例,往往会降低失败的可解释性。

shrinking 则适合做更局部的检查。 当我们怀疑某个 shrinker 的行为不稳定时,可以绕过随机生成,直接从一个具体值开始缩减:

///|
test "shrinking starts from explicit value" {
  let prop = @qc.shrinking(shrink_even_nat, 84, fn(x) { x < 20 })
  @qc.quick_check(prop, expect=Fail)
}

这类写法的妙处在于,它把「生成问题」和「缩减问题」拆开了。 如果一个性质本身已经能稳定失败,那么我们完全可以先固定一个已知失败值, 单独调试 shrinker 是否真的在朝着我们期望的业务极小值逼近。 这在设计复杂结构的 shrinker 时尤其有帮助,因为很多时候最难调的不是 generator, 而是「为什么最终反例还不够小」。当然,在实践中我们还是 forall_shrink 用的多, 或者使用 newtype 方式来用 trait 传参。

结构保持 Shrink​

到了受限结构场景,问题会进一步升级。 Part 2 我们已经看到,sorted array、BST、平衡树这类输入都带有明显的不变量。 如果 shrink 过程破坏了这些不变量,那么一个原本很好的失败样本, 就可能被缩成一个根本不该出现的非法输入,进而让失败原因变得模糊。

以 sorted array 为例,默认数组 shrink 会做两类事情:删除元素,以及缩小元素值。 这对普通数组很合理,但对「必须有序」的数组而言,第二类操作可能会破坏全局顺序。 并且使用过滤器的话会导致大量案例被浪费,效率也很差。 面对这种情况最好的思路就是直接在 shrink 过程中维护有序性不变量, 这需要我们自己编写 shrink 函数,来同时考虑「删除元素」和「缩小元素值」的合法性:

///|
pub fn[T : @qc.Shrink + Compare] shrink_sorted_array(
  xs : Array[T],
  lo~ : T,
  hi~ : T,
) -> Iter[Array[T]] {
  let shrink_one_val = (nv : Array[T]) => {
    let l = nv.length() - 1
    Array::makei(l + 1, i => i)
    .iter()
    .flat_map(i => {
      let lo = if i == 0 { lo } else { nv[i - 1] }
      let hi = if i == l { hi } else { nv[i + 1] }
      @qc.Shrink::shrink(nv[i]).flat_map(x => {
        if lo <= x && x <= hi && x != nv[i] {
          let nv1 = nv.copy()
          nv1[i] = x
          Iter::singleton(nv1)
        } else {
          Iter::empty()
        }
      })
    })
  }
  let remove_one_val = (v : Array[T]) => {
    let l = v.length() - 1
    Array::makei(l + 1, i => i)
    .iter()
    .flat_map(i => {
      let nv = v.copy()
      nv.remove(i) |> ignore
      Iter::singleton(nv)
    })
  }
  remove_one_val(xs).concat(shrink_one_val(xs))
}

这段代码体现了一个很重要的原则:对受限结构而言,「更小」并不足够, 「更小且仍然合法」才是我们真正需要的目标。 否则 shrink 虽然降低了字面规模,却把问题从「算法在合法输入上失败」 悄悄变成了「算法在非法输入上失败」,这种反例对定位 bug 的帮助会明显下降。

///|
test "shrink sorted array" {
  let s = shrink_sorted_array([1, 3, 5], lo=0, hi=9)
  inspect(
    s,
    content=(
      #|[[3, 5], [1, 5], [1, 3], [0, 3, 5], [1, 2, 5], [1, 3, 4], [1, 3, 3]]
    ),
  )
}

把这个 shrinker 接到性质测试上,就能得到一个始终保持有序性的缩减流程:

///|
test "forall_shrink for sorted array" {
  let gen = @qc.sorted_array(6, @qc.int_range(0, 9))
  let prop = @qc.forall_shrink(gen, x => shrink_sorted_array(x, lo=0, hi=10), fn(
    xs,
  ) {
    xs.length() < 3
  })
  let r = @qc.quick_check_silence(prop)
  inspect(
    r,
    content=(
      #|*** [0/0/100] Failed! Falsified.
      #|[0, 0, 0]
    ),
  )
}

这个性质本身非常简单,却足以说明结构保持 shrink 的意义。 生成器总是产生长度为 6 的有序数组,而 shrink 过程会继续尝试删除元素与缩小元素值, 直到逼近某个更小、但仍然有序的失败样本。 此处的 [0, 0, 0] 就是一个非常小的反例了,它直接指出了「有序数组长度限制」这个性质的核心问题。

同样的思想也适用于 Part 2 里的 BST 例子。 若 BST 是通过「数组 -> 插入 -> 树」这条路径构造出来的,那么更稳妥的 shrink 往往不是直接缩树节点, 而是回到其原始表示,先 shrink 数组,再重新执行 from_array 或 from_sorted。 这样做的好处是:生成与缩减共享同一份结构语义,不变量更容易维持,反例也更容易解释。

失败与分布​

失败信息​

缩减让我们拿到更小的反例,但一个更小的反例不一定就更好读。 很多时候真正困难的并不是「找到 bug」,而是「知道这个反例在业务语义里意味着什么」。 例如我们看到一个数组或树失败了,仍然可能不知道它失败前后的关键状态是什么, 也不知道失败究竟是由哪个派生条件触发的。 这时如果只打印原始输入,调试者仍然要把很多中间信息重新手算一遍。

QuickCheck 为此提供了 counterexample 这个组合子。 它并不会改变性质的真假,而只是把一段附加信息挂到失败输出上。 于是我们可以把「输入之外、但对解释失败很关键」的派生量, 例如中间结果、规格输出、归一化形式、路径标签等,一并打印出来。

///|
test "counterexample adds derived information" {
  let prop = @qc.forall(@qc.pure((0, [0, 0, -1])), fn(iarr) {
    let (x, arr) = iarr
    let out = remove_first_only(arr.copy(), x)
    @qc.counterexample(!out.contains(x), "after remove: \{out}")
  })
  let r = @qc.quick_check_silence(prop)
  inspect(
    r,
    content=(
      #|*** [0/0/100] Failed! Falsified.
      #|(0, [0, 0, -1])
      #|after remove: [0, -1]
    ),
  )
}

在这个例子里,原始输入 (0, [0, 0, -1]) 已经足够小, 但如果没有 after remove: [0, -1] 这条补充信息, 读者仍然需要自己执行一次函数,才能意识到「数组里还残留了一个 0」。

这种手法在模型对照与编译器测试里尤其重要。 我们常常会同时打印 expected / actual、归一化后的状态、 或者某段解释器轨迹的摘要信息。 当性质已经足够复杂时,失败输出本身其实就是一种局部调试报告, 而不是单纯的一组输入值。

分类统计​

失败信息解决的是单个反例如何解释的问题,而分类统计解决的是「整体样本分布长什么样」的问题。 即便一个性质一直通过,我们也不能立刻放心,因为生成器可能把大部分预算都花在了某一类平庸样本上, 或者根本没有覆盖到我们真正关心的分支。 如果这些分布信息不可见,那么所谓随机覆盖很容易退化成一种心理安慰。

QuickCheck 提供了 label、classify 与 collect 三个常用接口来观察测试数据。 label 适合给每个样本贴一个单独的标签, classify 适合把样本划入若干业务类别, collect 则更通用,它会把某个 Show 值直接当作标签收集起来。 在实践中,最常见的入口是先用 classify 观察几个关键类别是否真的出现过。

///|
fn t3_prop_rev_list(xs : @list.List[Int]) -> Bool {
  xs.rev().rev() == xs
}

///|
test "classify list distribution" {
  let r = @qc.quick_check_silence(
    @qc.Arrow(fn(xs : @list.List[Int]) {
      @qc.Arrow(t3_prop_rev_list)
      |> @qc.classify(xs.length() > 5, "long list")
      |> @qc.classify(xs.length() <= 5, "short list")
    }),
  )
  inspect(
    r,
    content=(
      #|+++ [100/0/100] Ok, passed!
      #|21% : short list
      #|79% : long list
    ),
  )
}

这个输出传达的信息并不在于性质通过了,而在于: 当前默认生成器明显更偏向较长的列表。 如果我们真正关心的是空列表、单元素列表、极短列表上的边界行为, 那这份统计就已经说明,单靠默认分布可能并不够,需要进一步调 generator。

label 与 collect 的用法也类似,只是粒度不同。 例如说我们可以直接用 label("length is \{xs.length()}") 观察每个长度桶的分布,也可以用 collect(xs.length()) 把长度这个量本身收集起来。 经验上,classify 更适合写教程和日常回归,因为类别更稳定、输出更容易读; 而 label / collect 更适合调生成器时做精细诊断。

Discard 分析​

分布问题里最容易被忽略的一类,是 discard。 当我们使用 @qc.filter、前置条件或者部分函数保护时, 很多样本可能会在真正执行性质之前就被丢弃。 少量 discard 是正常的,但如果 discard 数量持续偏高, 那往往说明我们把约束输入的工作推迟到了性质内部过滤的方法不适用。

先看一个温和的例子。我们只想测试非空列表,于是对默认生成器生成的列表再加一层过滤:

///|
test "discard on non-empty lists" {
  let prop_non_empty = fn(xs : @list.List[Int]) -> @qc.Property {
    (!xs.is_empty()) |> @qc.filter(!xs.is_empty())
  }
  inspect(
    @qc.quick_check_silence(@qc.Arrow(prop_non_empty)),
    content="+++ [100/40/100] Ok, passed!",
  )
}

这里测试虽然通过了,但我们仍然看到有 40 个样本被直接丢弃。 这意味着框架为了得到 100 个有效样本,实际上做了更多无用工作。 若这个前置条件再稀疏一些,浪费会迅速放大。

极端情况下,测试甚至会直接 gave up:

///|
test "reject all gives up" {
  let prop_reject = fn(_x : Int) { @qc.filter(true, false) }
  inspect(
    @qc.quick_check_silence(@qc.Arrow(prop_reject), expect=GaveUp),
    content="+++ [0/1000/100] Ok, gave up!",
  )
}

这个例子当然是刻意构造出来的,但它准确揭示了 discard 的语义: QuickCheck 并不是测试失败了,而是根本拿不到足够多的有效样本, 于是只能放弃。 因此一旦我们在真实项目里看到 gave up,或者发现 discard 数量长期偏高, 首先应该怀疑的不是性质本身,而是 generator 与前置条件之间是否存在结构性错位, 若遇到这种情况,第一步的修复往往是把约束条件尽量写进生成器,而不是写进过滤器。 这也正是为什么 Part 2 一直强调「尽量把约束写进生成器,而不是写进过滤器」。

SmallCheck​

小规模穷举​

到目前为止,我们主要讨论的还是 QuickCheck 风格的随机测试: 不断采样,再把失败样本通过 shrink 压小。 SmallCheck 走的是另一条路。 如果某类输入天然存在一个比较好的「从小到大」顺序, 那么我们就可以直接把这段前缀系统地测完。 它追求的不是概率覆盖,而是一个可复现、可穷尽、可解释的小规模搜索空间。

在 MoonBit QuickCheck 里,small_check 的入口和 quick_check 长得很像, 但它依赖的能力完全不同:

pub fn[A : @feat.Enumerable + Show, B : Testable] @qc.small_check(
  f : (A) -> B,
  max_size? : Int,
  expect? : Expected,
  abort? : Bool,
) -> Unit raise Failure

这里约束的不是 Arbitrary + Shrink,而是 Enumerable。 这意味着 SmallCheck 关心的不是「如何随机生成一个值」, 而是「如何把一个类型的值排成一个从小到大的顺序」。 只要这个顺序设计得当,测试就是确定性的:同样的 max_size 与性质,总会访问同样的前缀。

///|
test "small check fails on first non-zero int" {
  let r = @qc.small_check_silence(fn(x : Int) { x == 0 }, max_size=5)
  inspect(
    r,
    content=(
      #|*** [1/0/5] Failed! Falsified.
      #|1
    ),
  )
}

这个结果已经足以说明 SmallCheck 的工作方式。 对于 Int 的默认 Enumerable 实例,最前面的值依次是 0, 1, -1, 2, -2, ..., 所以性质 x == 0 会在第二个样本 1 处立即失败。 这也是为什么 SmallCheck 常常不需要额外 shrink:它一开始访问的就是小值前缀。

需要额外说明的是,经典 SmallCheck 往往以「深度上界」来描述其搜索范围; 而这里的实现更接近一个「枚举前缀」模型:max_size 控制本轮最多测试多少个值, 这些值来自 Enumerable 给出的有序枚举。 因此 SmallCheck 是否真正「从小到大」地覆盖了搜索空间, 本质上取决于 enumerator 的设计质量。

枚举器设计​

既然 SmallCheck 的核心是「枚举小值」,那么最重要的问题自然就变成: 什么才算一个设计良好的 enumerator。 从实现角度看,MoonBit 用 Enumerable trait 来承载这个信息:

pub(open) trait Enumerable {
  enumerate() -> @feat.Enumerate[Self]
}

pub fn[T] @feat.Enumerate::en_index(Self[T], BigInt) -> T

一个好的 enumerator 至少要满足三点。第一,枚举结果不应重复, 否则所谓「穷举前缀」就会浪费预算在同一个值上;第二,每个「大小层级」都必须是有限的, 否则 SmallCheck 会在某一层卡死,永远走不到更大的值; 第三,枚举顺序最好能反映我们对「复杂度」的直觉, 让越靠前的值越接近我们真正想要的「小样本」。

对递归数据类型而言,第二点尤其关键。 如果递归调用不额外增加任何代价,那么像自然数这样的类型会把无限多个值全塞进同一层, 从而破坏 part-finiteness,让我们的枚举进入无限循环。 因此 Feat 风格的枚举都会显式维护一个「付费」动作 pay, 表示每经过一层递归构造,值就进入下一层。

///|
enum Nat {
  Zero
  Succ(Nat)
} derive(Show, Eq)

///|
impl @feat.Enumerable for Nat with enumerate() {
  @feat.pay(fn() {
    @feat.singleton(Zero) +
    @feat.Enumerable::enumerate().fmap(fn(n) { Nat::Succ(n) })
  })
}

///|
test "nat enumerate order" {
  let e : @feat.Enumerate[Nat] = @feat.Enumerable::enumerate()
  let xs = [0N, 1, 2, 3, 4].map(fn(i) { e.en_index(i) })
  inspect(
    xs,
    content="[Zero, Succ(Zero), Succ(Succ(Zero)), Succ(Succ(Succ(Zero))), Succ(Succ(Succ(Succ(Zero))))]",
  )
}

这个定义看似很短,但已经体现了设计 enumerator 时最核心的原则。 @feat.singleton(Zero) 给出基例; 递归分支通过 fmap 把「更小的自然数」映射为 Succ(n); 而最外层的 pay 则保证每多套一层构造子,值就会被推迟到下一层。 如果去掉 pay,那么 Zero, Succ(Zero), Succ(Succ(Zero)), ... 就会落在同一个 part 里, SmallCheck 在理论上甚至无法完成这一层的枚举。

对更复杂的类型,写法也基本遵循同样的机械结构。 空构造器通常用 singleton; 单参数构造器可以直接 fmap; 多参数构造器则先借助 tuple/product 得到参数的枚举,再映射成真正的构造器; 若一个类型有多个构造器,就用 consts 或 union 把它们合并起来。 这里最关键的不是「把代码写短」,而是保持构造方式与数据语义一一对应, 确保枚举是无重复且按复杂度分层的。

Feat 风格​

上面的 Enumerable 接口并不是随意设计出来的,它基本上就是论文 《Feat: Functional Enumeration of Algebraic Types》中的「函数式枚举」思想在 MoonBit 里的落地。 它的核心观点很简单:与其把一个类型看成一条线性的值列表, 不如把它看成一组按大小分层的有限 parts。 每个 part 只关心两件事:这一层有多少值,以及如何按下标直接取值。

MoonBit 当前的实现也正是这样组织的。 Enumerate[T] 内部是一条惰性的 parts 序列,而每个 Finite[T] 则携带 fCard 与 fIndex 两个消费者。 于是全局索引 en_index 的语义就很清楚了: 它不是从头把所有值一个个生成出来,而是先根据各 part 的基数跳过整层, 再在命中的那一层里直接做索引。 这正是论文所谓的 function view,它和 SmallCheck 常见的 list view 有本质差异。

这样设计有两个直接好处。 其一,枚举不再局限于「从头扫到尾」,而是具备了随机访问能力。 其二,同一份 enumerator 可以同时支撑多种测试策略, 包括 SmallCheck 风格的前缀枚举,以及 @qc.Gen::feat_random 这类 size-bounded random sampling。 换句话说,Feat 并不是另一个单独的测试框架,而是一种共享的数据生成基础设施。

从论文角度看,Feat 对传统 SmallCheck 的修正也主要在这里。 经典 SmallCheck 更依赖构造深度这一概念,但深度并不总能准确刻画值的真实复杂度; 而 functional enumeration 会把「什么是小值」编码进 part 的构造过程里。 对于互递归 AST、语法树、类型系统中大量和与积交织的结构,这种分层通常比单纯的深度界更稳定, 也更容易被机械组合。

实践 SmallCheck​

理解了 enumerator 的设计之后,SmallCheck 的使用就会变得很自然: 我们先决定「什么叫小」,把这种次序写进 Enumerable, 再用 small_check 顺着这个顺序去检验前缀。 在这个过程中,测试质量很大程度上不再由随机种子决定,而由枚举顺序决定。

///|
test "small check on nat prefix" {
  let r = @qc.small_check_silence(fn(n : Nat) { n == Zero }, max_size=5)
  inspect(
    r,
    content=(
      #|*** [1/0/5] Failed! Falsified.
      #|Succ(Zero)
    ),
  )
}

这个例子与前面的整数版本形成了一个很好的对照。 由于 Nat 的 enumerator 是我们自己定义的, SmallCheck 所访问的「小值前缀」也完全由这份定义决定。 在这里,Zero 是第一个值,Succ(Zero) 是第二个值, 所以性质 n == Zero 会立刻在最小的非零自然数上失败。 这类反例几乎不需要后处理,因为「枚举顺序本身」已经承担了缩减的作用。

实际工程里,我们一般不会只为了找一个 Succ(Zero) 这样简单的反例去写 enumerator, 真正有价值的是更复杂的递归结构。 当一个类型包含多层递归、多个构造器,或者多组互递归定义时, QuickCheck 风格的手写 generator 往往会变得越来越像待测程序本身; 而 Feat 风格的 enumerator 通常仍然保持机械、局部、可组合。 这也是论文特别强调它适合大规模互递归语法树的原因。

总的来说,SmallCheck 很适合作为一种「前缀验证器」: 先系统扫过足够小、足够典型的值,尽快排除浅层错误; 如果这一前缀已经稳定通过,再把同一份 Enumerable 交给 feat_random 或其他随机策略,继续向更大的空间扩展。

总结​

到这里,这个 QuickCheck 教程系列的正文就算结束了。 回头看去,第一篇讲的是性质,第二篇讲的是输入, 这一篇补上的则是反例解释、失败诊断与更系统的小值覆盖。

QuickCheck、SmallCheck、functional enumeration 并不是非此即彼的关系。 输入空间太大时,随机生成与 shrink 往往是第一选择; 类型有自然的小值顺序时,SmallCheck 会更直接; 而当我们已经有了一份质量足够高的 Enumerable, 它也可以同时服务于穷举与随机。 真正需要避免的,不是选错工具,而是让性质、生成器与失败解释彼此脱节。

QuickCheck 教程 Part 2

· 阅读需 19 分钟

受限生成器的挑战​

基于属性测试(PBT)中最大的挑战之一是受限随机生成问题,现实场景往往不是简单的结构化生成器可以应对的, 我们可以自动 derive 简单类型的生成器,但是对于具有内部不变量的复杂类型,或者是受谓词约束的值, 此种手段就显得力不从心了。一个天真的想法是「先生成一个大范围的值,再在性质里过滤掉不满足条件的」, 但这会导致测试效率极低,甚至无法得到任何有效样本,因为满足条件的值分布往往非常稀疏。

考虑一个经典的受限 PBT 场景:

∀x,∀t,isBST(t)  ⟹  isBST(insert(t,x))\forall x,\forall t, \text{isBST}(t) \implies \text{isBST}(\text{insert}(t, x))

这表明,如果一棵树 tt 是有效的二叉搜索树(BST),那么在 tt 中插入一个新值 xx 会得到另一个有效的 BST。

为了测试这条性质,框架会反复采样 (x, t)。 如果 t 不是 BST,就会被直接丢弃; 只有通过前置条件的样本才会进入 insert 并检查结果。问题在于: 随机树成为 BST 的概率很低, 朴素生成器会让测试大部分轮次都浪费在 discard 上。 因此,在这种情况下我们可能不得不自己设计一个专门的生成器来直接生成满足 isBST 条件的树。 本文我们将由浅入深探讨自定义生成器的设计思路,并通过 QuickCheck 的 API 来实现它们,从而让我们能够高效地测试受限性质。

简单生成器​

我们先聚焦一类简单的生成器,它们是复杂生成器的基础构件, 主要负责对基础类型的值域进行约束、混合容器类型、积类型等等。

值范围控制​

在 PBT 中,最常用的起点是对基础类型的取值范围进行建模, 更确切的说,是对与有序类型的值域进行约束。对于整数,我们可能只关心某个区间内的值;对于字符,我们可能只关注特定范围或类别的字符。 在 QuickCheck 中,我们有 @qc.int_range、@qc.small_int、@qc.nat、 @qc.neg_int 这些函数用于表达不同的整数域, 也有 @qc.char_range、@qc.alphabet、@qc.numeral 用于字符域的约束表达。 我们通常先用这些生成器把输入限制在需求语义允许的范围内,再由性质去验证更高层的关系。

///|
test "gen @qc.int_range invariant" {
  let gen = @qc.int_range(-10, 10)
  let prop = @qc.forall(gen, fn(x) { x >= -10 && x <= 10 })
  @qc.quick_check(prop)
}

生成器并不总是「随机」的,我们也可以通过 @qc.pure 构造一个恒定值生成器,用来表达边界场景或固定前置条件。 这类生成器在组合时非常重要,它们能稳定地把某些输入固定住,从而让我们聚焦于另一部分输入的变化。

///|
test "gen @qc.pure value" {
  let gen = @qc.pure(7)
  let prop = @qc.forall(gen, fn(x) { x == 7 })
  @qc.quick_check(prop)
}

Arbitrary 导出生成器​

Arbitrary 是 QuickCheck 中一个重要的 trait,它定义了如何为某个类型生成随机值。 MoonBit 编译器内置了 Arbitrary 实例的自动派生机制,能够为大多数简单类型生成默认的随机值。 只需要直接写 derive(Arbitrary) 即可。

///|
enum Color {
  Red
  Green
  Blue
} derive(Arbitrary, Show)

如果我们已经为某个类型定义了 Arbitrary 实例, 那么 @qc.Gen::spawn 可以直接生成默认分布的生成器。 它与 @qc.quick_check_fn 的隐式生成逻辑一致,但允许我们显式地插入到 @qc.forall 之中, 从而在组合生成器时保持结构清晰,且能够继续叠加其他约束。

///|
test "gen spawn for arbitrary" {
  let gc : @qc.Gen[Color] = @qc.Gen::spawn()
  let gen : @qc.Gen[Int] = @qc.Gen::spawn()
  inspect(
    gc.samples(size=5),
    content=(
      #|[Green, Green, Green, Blue, Red]
    ),
  )
  inspect(
    gen.samples(),
    content=(
      #|[6, 4, -6, -3, 0, 2, -8, 4, 5, 2]
    ),
  )
}

集合结构与多参组合​

当需求涉及集合结构时,基础生成器需要能够表达「长度」与「元素来源」。 @qc.Gen::array_with_size 提供了固定长度数组的生成能力, @qc.list_with_size 则用于构造指定长度的列表。固定长度并非只为了便于测试, 它往往直接对应了协议、格式或算法的前提条件。

///|
test "gen array_with_size" {
  let gen = @qc.int_range(0, 9).array_with_size(5)
  json_inspect(gen.samples(size=5), content=[
    [0, 6, 4, 3, 8],
    [0, 4, 6, 5, 7],
    [5, 2, 0, 0, 2],
    [4, 1, 0, 4, 3],
    [5, 3, 3, 1, 4],
  ])
}

///|
test "gen @qc.list_with_size sample" {
  let gen = @qc.char_range('a', 'f').list_with_size(3)
  json_inspect(gen.sample(), content=["a", "b", "f"])
}

多参数函数是实际业务的常态, 而 @qc.tuple、@qc.triple、@qc.quad 让我们可以将多个生成器合成为一个输入, 从而保持「单参性质」的统一执行模型。这样做不仅让性质更简洁, 也让缩减过程能够同时关注多个参数之间的相互作用。

///|
test "gen @qc.tuple for two args" {
  let gen = @qc.tuple(@qc.int_range(-20, 20), @qc.int_range(-20, 20))
  let prop = @qc.forall(gen, fn(p) {
    let (a, b) = p
    a - b + b == a
  })
  @qc.quick_check(prop)
}

基础结构的最后一个关键环节是「变换」。@qc.Gen::fmap 允许我们在生成结果之上进行纯函数变换, 从而把已有的值域映射为新的域。这个能力看似简单,却是我们构造业务特化输入的核心手段, 后续的分布控制与条件过滤也会建立在这一层结构之上。

///|
test "gen fmap transform" {
  let gen = @qc.int_range(0, 50).fmap(fn(x) { x * 2 })
  let prop = @qc.forall(gen, fn(x) { x % 2 == 0 })
  @qc.quick_check(prop)
}

通过这些基础结构,我们已经可以覆盖大量现实需求中最常见的输入形态:受限数值、固定长度集合以及多参组合。 在此基础上,我们下一步要解决的是「分布是否合理」的问题,也就是如何在可控的前提下更接近真实数据形态, 这将是下一章的核心内容。

统计分布控制​

有了「合法形状」的输入后,下一个问题是: 这些输入出现得是否像真实世界一样频繁? 这就需要我们控制分布。现实数据往往呈现多峰、偏态或结构性特征,若只依赖单一范围生成器, 测试覆盖会显得单薄。我们需要通过组合与加权,让输入分布更贴近真实场景,同时保持性质表达的简洁性。

当需求存在多种类别或路径时,@qc.one_of 是最直接的组合手段。它在若干生成器之间做均匀选择, 适合把边界样本与常规样本并置,让性质既能触及极端情况,也能覆盖正常区间的变化。

///|
test "gen @qc.one_of mix" {
  let gen = @qc.one_of([@qc.pure(0), @qc.pure(1), @qc.int_range(-10, 10)])
  let prop = @qc.forall(gen, fn(x) { x >= -10 && x <= 10 })
  @qc.quick_check(prop)
}

均匀选择在很多场景并不理想,现实数据往往有明显的主流区间或热点值。此时我们可以使用 @qc.frequency 对分支赋权,表达「多数情况来自某个范围,少数情况来自另一个范围」的分布设计,从而把测试资源集中到 更可能出错的区域,同时仍保留稀有路径的覆盖。

///|
test "gen @qc.frequency weighted" {
  let gen = @qc.frequency([
    (6, @qc.int_range(-3, 3)),
    (1, @qc.int_range(-30, 30)),
  ])
  let prop = @qc.forall(gen, fn(x) { x >= -30 && x <= 30 })
  @qc.quick_check(prop)
}

对离散枚举值而言,@qc.one_of_array 与 @qc.one_of_list 更加自然, 它们直接从给定集合中取值,避免构造过度复杂的生成器。 我们通常用它来模拟协议字段、状态码或固定集合的配置值,从而使性质更接近真实输入。

///|
test "gen @qc.one_of_array enum" {
  let methods : Array[String] = ["GET", "POST", "PUT"]
  let gen = @qc.one_of_array(methods)
  let prop = @qc.forall(gen, fn(m) { methods.contains(m) })
  @qc.quick_check(prop)
}

当多个字段存在依赖关系时,@qc.Gen::bind 可以将这种依赖编码进生成阶段。它允许我们先生成一个值, 再根据该值生成后续字段,从而在数据层面满足约束,避免在性质内部叠加大量前置判断。

bind 是非常强大的单子操作,它让我们能够在生成过程中动态地调整分布与结构,从而直接生成满足复杂关系的输入。 当然它也更加难以理解与调试,因此我们需要在使用时保持清晰的层次结构,避免过度嵌套或过度依赖 bind 来表达复杂逻辑,

///|
test "gen bind dependent" {
  let gen = @qc.int_range(-10, 10).bind(fn(base) {
    @qc.int_range(0, 5).fmap(fn(delta) { (base, base + delta) })
  })
  let prop = @qc.forall(gen, fn(p) {
    let (a, b) = p
    a <= b && b - a <= 5
  })
  @qc.quick_check(prop)
}

此前已经介绍过了 @qc.Gen::fmap ,它仍然是组合中的基础能力,可在不改变分支概率的前提下,把生成结果映射为业务结构。 这种映射保持了分布的形状,却让数据更贴合接口语义,因而常被用于构造标识符、规范化输入或衍生字段。

在实践中,我们通常先用 @qc.one_of 或 @qc.frequency 设定「宏观分布」,再用 bind 与 fmap 完成 「微观结构」的约束与衍生。这样的两层结构能够同时兼顾覆盖面与真实性,并且保持生成器的可读性。 组合与分布并不会改变性质本身,但会显著影响测试的有效性。分布设计应当围绕需求语义展开,避免过度平均, 也避免过度偏置,从而使随机测试在有限预算内提供更可靠的缺陷发现能力。

在此基础之上,我们还需要进一步控制规模与复杂度,这涉及 size 参数的演化与生成器的尺度策略, 这将是下一章讨论的重点。

规模与复杂度​

本章讨论 size 参数如何影响数据规模与测试复杂度。随机测试并不是越大越好,规模过大会掩盖问题本质, 规模过小又会失去覆盖价值。我们需要把 size 当作成本与收益之间的调节器,通过配置与生成器策略让测试 在可控的预算内逼近真实复杂度。

@qc.quick_check 提供了 max_size 用于限制整体规模,这是最直接的控制方式。我们常在算法复杂度较高、 或输入域可能指数增长的场景中使用它,以避免测试时间失控,同时确保性质仍能在合理范围内被充分检验。

///|
test "@qc.quick_check max_size" {
  let gen = @qc.sized(fn(n) { @qc.small_int().list_with_size(n) })
  let prop = @qc.forall(gen, fn(xs) { xs.length() >= 0 })
  @qc.quick_check(prop, max_size=30)
}

当我们希望「数据结构与 size 同步增长」时,@qc.sized 是更明确的表达。它将 size 作为参数传入生成逻辑, 从而把规模约束写在生成器内部,避免在性质中处理尺寸相关的前置条件。这种方式对数组、列表、树等结构 尤其有效,因为它将复杂度控制内化为输入域的构造规则。

///|
test "@qc.sized array with explicit length" {
  let gen = @qc.sized(fn(n) {
    let len = if n < 0 { 0 } else { n }
    @qc.tuple(@qc.pure(len), @qc.int_range(0, 9).array_with_size(len))
  })
  inspect(
    gen.sample(),
    content=(
      #|(100, [5, 5, 0, 2, 0, 5, 4, 6, 4, 2, 1, 3, 3, 3, 0, 8, 2, 4, 2, 3, 5, 6, 5, 8, 8, 6, 2, 1, 7, 3, 6, 6, 1, 3, 8, 3, 4, 7, 4, 8, 7, 4, 0, 7, 2, 5, 4, 6, 5, 5, 8, 8, 5, 6, 5, 6, 2, 3, 5, 7, 3, 3, 0, 3, 7, 4, 0, 4, 0, 7, 6, 6, 2, 5, 1, 5, 3, 3, 2, 7, 8, 8, 8, 1, 4, 2, 8, 0, 8, 8, 4, 2, 6, 5, 0, 2, 5, 2, 0, 6])
    ),
  )
}

当我们需要在不修改生成器结构的前提下限制规模时,可以使用 @qc.Gen::resize。它会把 size 固定为指定值, 从而将复杂度稳定在一个可预期的水平。我们常在调试或回归阶段使用它,让反例更集中、运行时间更稳定。

///|
test "resize clamps size" {
  let gen = @qc.sized(fn(n) { @qc.int_range(0, 9).list_with_size(n) })
  let small = gen.resize(5)
  let prop = @qc.forall(small, fn(xs) { xs.length() == 5 })
  @qc.quick_check(prop)
}

如果我们希望规模随 size 变化,但增长速度不那么陡峭,则可以使用 @qc.Gen::scale 调整 size 的映射关系。 这相当于在「生成复杂度曲线」上加一层函数,使数据规模随测试轮次增长得更平缓,从而在有限预算中 获得更稳定的覆盖与更可控的运行时间。

///|
test "scale slows growth" {
  let gen = @qc.sized(fn(n) { @qc.int_range(0, 9).list_with_size(n) })
  let scaled = gen.scale(fn(n) { n / 2 })
  let prop = @qc.forall(scaled, fn(xs) { xs.length() <= 20 })
  @qc.quick_check(prop, max_size=40)
}

规模控制不仅影响性能,也影响失败解释。过大的结构会导致缩减时间增加,反例噪声加重,甚至掩盖关键路径。 因此我们需要把 max_size、resize 与 scale 作为统一策略使用,在不同阶段选择不同的规模曲线, 让性质既能触及复杂情形,又能保持失败信息的可读性与可诊断性。

组合构造器​

到这里我们已经有了范围、分布与规模控制,剩下的难点是「结构性约束」。 例如二分查找、区间合并等逻辑都要求输入数组有序,单靠 int_range 这类基础生成器无法直接表达这个前置条件。 一个直接做法是先生成数组,再用 @qc.filter 过滤为有序样本,这正是组合构造器最常见的入口。

///|
test "combinator sorted array with filter" {
  fn is_non_decreasing(xs : Array[Int]) -> Bool {
    fn go(i : Int) -> Bool {
      if i + 1 >= xs.length() {
        true
      } else {
        xs[i] <= xs[i + 1] && go(i + 1)
      }
    }

    go(0)
  }

  let base = @qc.int_range(-8, 8).array_with_size(3)
  let prop = @qc.forall(base, fn(arr) {
    @qc.forall(@qc.one_of_array(arr), fn(x) {
      arr[0] <= x && x <= arr[arr.length() - 1]
    })
    |> @qc.filter(is_non_decreasing(arr))
  })

  @qc.quick_check(prop, discard_ratio=20)
}

这个例子里有三个组合层次:先用 array_with_size 固定结构,再用嵌套 forall + one_of_array 建立元素与容器的依赖, 最后用 filter 施加「有序」约束。写法直观,适合快速验证想法,但它仍然会丢弃一部分样本。

当过滤比例偏高时,我们更推荐把约束提前到「构造阶段」。 QuickCheck 已经提供 @qc.sorted_array,我们可以直接利用它:

///|
test "combinator sorted array constructor" {
  let gen = @qc.sorted_array(5, @qc.int_range(-30, 30))
  let prop = @qc.forall(gen, fn(arr) {
    @qc.forall(@qc.one_of_array(arr), fn(x) {
      arr[0] <= x && x <= arr[arr.length() - 1]
    })
  })
  @qc.quick_check(prop)
}

filter 适合表达「临时前置条件」,sorted_array 这类构造器适合表达「稳定结构不变量」。 在工程实践中通常先用过滤器快速定位性质,再逐步替换为专门构造器,让测试既可读又高效。

受限结构的构造器案例​

在介绍完上述的各种组合子之后,是时候进入我们的正题: 设计一个满足特定性质的生成器,例如有序数组、平衡树、特定协议格式等。 当然,这并非易事,本节也只能提供一个大致的思路框架, 复杂情况仍然需要测试者进行创造性的设计与调试。

手写 QuickCheck 生成器的核心目标其实就两件事:

  • 把「有效输入空间」编码进生成过程,别靠过滤;
  • 让分布与规模(size)可控,这样测试既跑得动又能覆盖到你真正关心的结构角落

规模作为一等公民​

QuickCheck 的 Gen 有一个隐含的 size 参数: 随着测试次数增长, size 会逐步变大。手写递归结构生成器时, 最重要的是每一层递归都要消耗 size,否则要么无限递归, 要么结构大得离谱导致测试变慢。

fn gen_t() -> @qc.Gen[T] {
  letrec go = (s : Int) => {
    match s {
      0 => base case
      n => recursive case, can call go(n - 1) for smaller substructures
    }
  }
  @qc.sized(go)
}

如果你不想「每层都减 1」, 也可以把 n 拆成左右子规模 (树、图、AST 都是这个套路): let k = int_range(0, n - 1), 左用 k,右用 n-1-k。这样平均会更自然, 而不是总在深度上单边生长,当然也可以直接 go(n / 2),让规模指数级增长。

二叉搜索树的例子​

先定义 BST 的数据结构:

///|
enum Tree[T] {
  Leaf
  Node(Tree[T], T, Tree[T])
} derive(Debug, Show)

如果性质测试不强依赖「树形分布」, 第一个方案是,我们可以先定义一个「插入」函数,来把任意值插入到 BST 中, 然后用 from_array 来把一个数组转成 BST。 这样我们就能直接利用 @qc.int_range().array_with_size() 来先生成一个普通数组, 再通过 from_array 来得到一棵树, 这样天然满足 BST 不变量, 而且 shrink 也很好做(缩列表即可)。

///|
fn[T : Compare] Tree::insert(t : Tree[T], v : T) -> Tree[T] {
  match t {
    Leaf => Node(Leaf, v, Leaf)
    Node(l, x, r) if v < x => Node(l.insert(v), x, r)
    Node(l, x, r) if v > x => Node(l, x, r.insert(v))
    Node(l, x, r) => Node(l, x, r) // v == x, no duplicates
  }
}

///|
fn[T : Compare] Tree::from_array(arr : Array[T]) -> Tree[T] {
  arr.fold(init=Leaf, Tree::insert)
}

///|
/// inorder traversal should be sorted
fn[T] inorder(tree : Tree[T]) -> Array[T] {
  match tree {
    Leaf => []
    Node(l, x, r) => inorder(l) + [x] + inorder(r)
  }
}

///|
test "generate BST" {
  let int_arr = @qc.int_range(-100, 100).array_with_size(10)
  let gen_bst = int_arr.fmap(Tree::from_array)
  let prop = @qc.forall(gen_bst, fn(t) {
    let arr = inorder(t)
    arr == arr.copy()..sort()
  })
  @qc.quick_check(prop)
}

然而,这需要我们理解 BST 的插入逻辑,才能正确地实现 Tree::insert, 如果这个函数也实现错误,那我们的测试结果会非常混乱。 并且这个方案的效率也不高,因为 from_array 可能会生成非常不平衡的树, 因此我们的下一个优化目标可能是「让树更加平衡」,从而更快地覆盖到不同的树形结构。

BST 的一个天然表示是中序遍历是有序序列, 所以我们可以先生成一个列表,排序去重, 然后用「取中点」方式建近似平衡树:

///|
fn[T] from_sorted(arr : ArrayView[T]) -> Tree[T] {
  guard !arr.is_empty() else { Leaf }
  let m = arr.length() / 2
  let (l, x, r) = (arr[:m], arr[m], arr[m + 1:])
  Node(from_sorted(l), x, from_sorted(r))
}

///|
test "generate balanced BST" {
  let int_arr = @qc.int_range(-100, 100).array_with_size(10)
  let gen_bst = int_arr.fmap(fn(arr) { arr..sort()..dedup() |> from_sorted })
  let prop = @qc.forall(gen_bst, fn(t) {
    let arr = inorder(t)
    arr == arr.copy()..sort()
  })
  @qc.quick_check(prop)
}

这个方案的优势是:大多数树更平衡, 能更容易覆盖到「左右子树都非空」的情形, 也能减小极端深度带来的性能问题。

下一个方案叫做区间递归生成,直接按照 BST 的语义来生长, 关键点是在递归时维护值域区间 (lo, hi): 左子树只能取 (< root),右子树只能取 (> root)。 这能控制很多细节,下面我们以 Tree[Int] 为例 (因为 QuickCheck 天然提供了 int_range):

///|
fn gen_bst_ranged(min : Int, max : Int) -> @qc.Gen[Tree[Int]] {
  letrec go = (n : Int, lo : Int, hi : Int) => {
    guard lo <= hi && n > 0 else { @qc.pure(Leaf) }
    @qc.frequency([
      (1, @qc.pure(Leaf)),
      (
        4,
        @qc.Gen::new((i, rs) => {
          let x = @qc.int_range(lo, hi).run(i, rs)
          let nL = @qc.int_range(0, n - 1).run(i, rs)
          let nR = n - 1 - nL
          let l = go(nL, lo, x - 1).run(i, rs)
          let r = go(nR, x + 1, hi).run(i, rs)
          Node(l, x, r)
        }),
      ),
    ])
  }
  @qc.sized(n => go(n, min, max))
}

///|
test "generate ranged BST" {
  let gen_bst = gen_bst_ranged(-100, 100)
  let prop = @qc.forall(gen_bst, fn(t) {
    let arr = inorder(t)
    arr == arr.copy()..sort()
  })
  @qc.quick_check(prop)
}

注意这里为了避免重复, 用了 x-1/x+1 这样的「离散域」技巧; 如果你允许重复值, 就要改成 l 用 (lo, x),r 用 (x, hi) 并决定重复放哪边(<= 或 >= 的约定必须统一)。

区间法的真实价值在于: 当你的结构约束更复杂(比如红黑树、AVL、带额外标签的 AST), 你可以把「约束状态」一路带下去,让生成永远有效,而不是靠过滤赌概率。 换句话说,这种方法永远是最具扩展性的,因为它把「语义不变量」直接编码在生成逻辑里了。

总结​

受限生成器的设计是 PBT 中的核心挑战之一。 通过合理地组合基础生成器、控制分布与规模,并把语义约束内化到生成过程中, 确实可以满足大多数受限性质的测试需求。当然,未来我们希望让它更加自动化, 例如用户只需要编写性质,就可以推导出满足性质前置条件的生成器, 从而让 PBT 的使用门槛更低,覆盖更广。 这是相当具有可行性的,我们将在下一篇文章更多讨论这些前沿技术 (inductive relations / functional enumeration 等等)。

QuickCheck 教程 Part 1

· 阅读需 22 分钟

这是一个长系列,目标是系统介绍 MoonBit 中的 QuickCheck 框架及其在工程实践中的应用, 向广大开发者介绍「基于属性的测试设计」理念与方法论。 整个系列将分为 3 个大部分,第一部分,也就是本文, 关注 QuickCheck 的核心概念和『属性』的设计方法论, 后面部分将介绍生成器的高级设计与缩减策略和统计分布控制等技巧。

基本概念解释​

本章旨在建立我们对「性质测试」的直观理解,我们不急着讨论复杂生成器或者属性, 而是从最直观的角度确认 QuickCheck 在 MoonBit 中到底验证了什么。 这里的性质 (property) 不是一组样例,而是一段描述规则的程序, 它会被反复运行在大量随机数据上(在程序上可以体现为一个函数或者一个可执行的表达式), 从而让我们以更低成本覆盖更大行为空间。

当我们把一个性质交给 QuickCheck 时,框架需要先判断它是不是可执行的,为此 QuickCheck 提供了 Testable 这一抽象,它能把 Bool、Function、甚至 Generator 统一成可运行的 Property。换句话说,我们写的不一定是 「测试」,而是一个能被转译成测试流程的值,这个值会被运行、统计、收集并最终给出结论。

fn[P : @qc.Testable] @qc.quick_check(prop : P, max_success? : Int, ...) -> Unit raise Failure

上面的函数签名展示了 QuickCheck 的核心入口 @qc.quick_check。它接受一个 Testable 值,将其转成 Property 并运行测试。max_success 参数控制测试样例数量,默认值为 100。若性质在所有样例上成立,测试通过;否则抛出 Failure 并打印反例,当然它不止 max_success 这个可配置参数,之后的章节中,我们将按需引入不同的配置项。

///|
test "@qc.quick_check minimal" {
  @qc.quick_check(true)
}

这个最小示例几乎没有任何信息,却能帮助我们准确把握接口形态。 @qc.quick_check 接受任何 Testable 值, 因此布尔值也可以直接作为性质 (因为它实现了这个 trait), 若为 false 就会失败。这样的调用虽然极简,但清晰揭示了 QuickCheck 的入口语义,即它只关心「这段可执行的性质最终是否成立」。

当性质是一个函数时,最便捷的入口是 @qc.quick_check_fn。 它要求函数的参数类型具备 Arbitrary、Shrink 与 Show 能力, 从而能自动生成测试数据、缩减反例并打印失败样本,在后面我们会更多解释这些 trait 的含义, 但现在你只需要知道对于标准库的基础类型,QuickCheck 都实现了这些 trait。 我们可以把它理解为「我们提供规则,系统替我们提供数据」的测试模式,在简单模型上非常高效。 下面的例子也很简单,验证了「整数加零不变」的性质,生成器会从小到大生成 100 个整数样例并运行该性质, 并检查结果是否全部为真,如果是,则测试通过,否则打印出第一个失败的样例。

///|
fn prop_add_zero(x : Int) -> Bool {
  x + 0 == x
}

///|
test "@qc.quick_check_fn" {
  @qc.quick_check_fn(prop_add_zero)
}

实际业务函数往往是多参的,而 property 又只能接收一个参数,因此我们用元组把多个输入打包起来。 这不是权宜之计,而是语言层面的标准做法,它让 QuickCheck 仍然保持「单参性质」的统一执行流程, 同时也便于缩减时生成更小的反例组合。

///|
fn prop_add_comm(pair : (Int, Int)) -> Bool {
  let (a, b) = pair
  a + b == b + a
}

///|
test "@qc.tuple property" {
  @qc.quick_check_fn(prop_add_comm)
}

当我们需要主动决定数据分布时,或者 Arbitrary 实例并不适用时, 就需要显式生成器,这个入口是 @qc.forall。@qc.forall 的含义是「对生成器给出的每一个值, 都让性质成立」,它把生成器与性质函数粘合成 Property,是后续复杂设计的基础。 在这里我们先给出一个小范围的整数生成器,让性质更容易直观地理解。

///|
test "@qc.forall with @qc.int_range" {
  let gen = @qc.int_range(-10, 10)
  let prop = @qc.forall(gen, fn(x) { x + 1 > x })
  @qc.quick_check(prop)
}

生成器并非神秘黑箱,它是一个由 size 与随机种子驱动的确定性函数。 虽然 @qc.quick_check 会帮我们自动管理这些参数,但在设计性质时, 我们仍然可以用 @qc.Gen::sample 先窥视生成器的行为,帮助我们校准 数据分布是否符合预期。

///|
test "peek generator" {
  let gen = @qc.int_range(-3, 3)
  inspect(gen.sample(size=5, seed=1), content="-2")
}

在这里我们可以讨论一些 QuickCheck 内部的结构细节:

从执行流程上看,Testable 会被转成 Property,Property 内部再被展开成可遍历的测试树,QuickCheck 沿着这棵树运行、记录、缩减并最终决定结果。我们暂时不需要理解这些结构细节,但要清楚性质是「可执行」 的对象,而不是静态文档,这一层理解会决定我们之后如何组织属性与生成器。

当然读者无需担心这些细节会妨碍我们使用 QuickCheck,框架已经帮我们封装好了这些复杂性, 我们只需专注于「写性质」与「选生成器」即可。

失败的处理也是接口语义的一部分,@qc.quick_check 会在性质失败时抛出 Failure 并打印反例,而 @qc.quick_check_silence 则返回一段报告字符串,方便我们在工具链中做二次处理。理解这一点有助于我们在 调试和持续集成中选择合适的入口。

///|
test "@qc.quick_check_silence" {
  let prop = @qc.forall(@qc.int_range(0, 5), fn(x) { x >= 0 })
  inspect(@qc.quick_check_silence(prop), content="+++ [100/0/100] Ok, passed!")
}

到这里我们已经能把一个直观规则写成可运行的性质,并通过默认生成器或显式生成器运行它。我们会发现 QuickCheck 并不要求我们先理解复杂的缩减细节,而是提供了一条从「规则」到「执行」的清晰通路, 这也是 property-based testing 真正降低测试成本的原因。

Property 设计导论​

本章关注如何从需求/代码中抽取可验证的性质,我们不急于讨论复杂生成器, 而是把注意力放在「关系」与「不变量」上。 需求常以自然语言或数学公司表达,它蕴含了不变量与代数规律, 我们要做的就是把这些规律转化为可执行的 property, 并让随机测试去检验它们是否稳定成立。

代数性质​

在性质测试中,最可靠的起点是「关系式」,它描述输入与输出之间应当长期成立的约束。相比样例断言, 关系式具有普适性,能够覆盖更大范围的输入组合。我们可以把「应该相等」、「应当保持顺序」或 「重复应用后不再变化」这样的语义转化为函数层面的规律,并将其交给 QuickCheck 执行。

QuickCheck 收集了常见的代数规律作为内置性质,这些规律直接对应了需求中常见的模式。例如, 当需求暗含交换性时,我们可以直接采用 @qc.commutative 来表达等式关系。这里我们用整数加法作为示例, 并限制输入范围以避免溢出干扰性质本身。这样的范围设置不是削弱测试,而是帮助我们聚焦在需求语义上。

///|
test "@qc.commutative for add" {
  let gen = @qc.tuple(@qc.int_range(-200, 200), @qc.int_range(-200, 200))
  let prop = @qc.forall(gen, @qc.commutative(fn(a, b) { a + b }))
  @qc.quick_check(prop)
}

需求中常见的「合并不依赖分组方式」可以用结合律表达,这类规律尤其适用于聚合、拼接、合并等函数。 我们借助 @qc.associative 直接表达结合律,并用三元组生成器将多参输入统一为单参性质。

///|
test "@qc.associative for add" {
  let gen = @qc.triple(
    @qc.int_range(-20, 20),
    @qc.int_range(-20, 20),
    @qc.int_range(-20, 20),
  )
  let prop = @qc.forall(gen, @qc.associative(Int::add))
  @qc.quick_check(prop)
}

当需求包含「分配」的语义时,我们通常需要把两个运算的关系固定下来。分配律不仅揭示了运算组合的结构, 也能快速检验实现是否正确地遵守数学规则。我们在这里选用乘法对加法的左分配律来表达这一类需求。

///|
test "@qc.distributive_left for mul/add" {
  let gen = @qc.triple(
    @qc.int_range(-12, 12),
    @qc.int_range(-12, 12),
    @qc.int_range(-12, 12),
  )
  let prop = @qc.forall(gen, @qc.distributive_left(Int::mul, Int::add))
  @qc.quick_check(prop)
}

除了代数律,很多业务需求本质上是「重复应用不会继续改变结果」。这类需求适合用幂等性刻画, 如归一化、去噪、裁剪等操作。我们可以先写出一个简单的非负裁剪函数,再用 @qc.idempotent 验证其性质。

///|
fn clamp_nonneg(x : Int) -> Int {
  guard x < 0 else { x }
  0
}

///|
test "@qc.idempotent clamp" {
  let prop = @qc.forall(@qc.int_range(-50, 50), @qc.idempotent(clamp_nonneg))
  @qc.quick_check(prop)
}

另一个常见模式是「反演回到原处」,也就是自反或对合性质。许多编码与解码、加密与解密、开关与还原 都可以用这一结构来建模。我们在此用取负作为最简示例,并用 @qc.involutory 表达「再应用一次即可回原值」。

///|
test "@qc.involutory neg" {
  let prop = @qc.forall(@qc.int_range(-100, 100), @qc.involutory(Int::neg))
  @qc.quick_check(prop)
}

当需求描述的是「不同实现应给出相同结果」时,@qc.ext_equal 是更直接的表达。它并不关心内部算法, 而只要求两个实现对所有输入给出相同输出,这一点非常适合重构或优化后的回归验证。

///|
fn double1(x : Int) -> Int {
  x + x
}

///|
fn double2(x : Int) -> Int {
  x * 2
}

///|
test "@qc.ext_equal for double" {
  let prop = @qc.forall(
    @qc.int_range(-100, 100),
    @qc.ext_equal(double1, double2),
  )
  @qc.quick_check(prop)
}

有些需求体现的是「可逆」的含义,这时我们也可以用 @qc.inverse 来描述它。我们通过一个增量和减量函数 来表达可逆性,并在受控的输入范围内验证这一关系。注意这里我们仍用生成器限制输入,避免超出语义前提。

///|
fn inc(x : Int) -> Int {
  x + 1
}

///|
fn dec(x : Int) -> Int {
  x - 1
}

///|
test "@qc.inverse for inc/dec" {
  let prop = @qc.forall(@qc.int_range(-100, 100), @qc.inverse(inc, dec))
  @qc.quick_check(prop)
}

从这些例子可以看到,性质设计的关键并不在于「写更多断言」,而在于选择正确的结构来表达需求。 当我们将需求映射为代数规律或等价关系时,测试就不再是碎片化的样例,而是对整个输入空间的系统抽样。 此外,性质并非越强越好。过强的性质可能隐含不真实的假设,过弱的性质又难以约束实现。 因此我们在设计时需要让性质可解释、可证伪,同时尽量与需求文本保持可追溯的对应关系。

易错点​

代数性质很常见,但它并非银弹,在设计时更需注意以下易错点:

  • 部分函数:当性质涉及除零、下标越界等部分函数时,需确保生成器避免这些输入,或在性质中处理异常情况。
  • 浮点数:浮点数的精度与特殊值(NaN、Infinity)可能导致性质失效,需谨慎设计生成器与性质逻辑,并且不应该 直接用等式比较浮点数,考虑误差界 ∣x−y∣<ϵ\mid x - y \mid < \epsilon。
  • 分布问题:性质设计时需考虑生成器的分布是否合理,过于均匀或偏态的分布可能导致测试覆盖不足, 在后面的生成器章节我们将详细讨论这一点。

操作不变量​

本节讨论为何「仅测试公理」在抽象数据类型的语境中可能产生误判的经典情况, 以及如何通过操作不变性测试来弥补这一缺口。我们尤其关注那些隐藏了内部表示的抽象数据类型, 用户只能通过公开操作观察行为。 举个例子,我们考虑 FIFO 队列,给出操作签名与公理 Q1–Q6,并展示了一个具有内部不变量的实现。

///|
declare fn empty() -> Queue

///|
declare fn enqueue(x : Int, q : Queue) -> Queue

///|
declare fn is_empty(q : Queue) -> Bool

///|
declare fn front(q : Queue) -> Int

///|
declare fn dequeue(q : Queue) -> Queue

公理:

  • Q1: is_empty(empty()) == true
  • Q2: is_empty(enqueue(x, q)) == false
  • Q3: front(enqueue(x, empty())) == x
  • Q4: !is_empty(q) => front(enqueue(x, q)) == front(q)
  • Q5: dequeue(enqueue(x, empty())) == empty()
  • Q6: !is_empty(q) => dequeue(enqueue(x, q)) == enqueue(x, dequeue(q))

一个很天真的测试方法就是写好 Queue 之后我们直接使用这组签名进行测试:

///|
/// `gen_queue()` 是一个生成随机 Queue 实例的生成器
test "property Q2" {
  let prop = @qc.forall(@qc.tuple(@qc.small_int(), gen_queue()), fn(p) {
    let (x, q) = p
    is_empty(enqueue(x, q)) == false
  })
  @qc.quick_check(prop)
}

这些测试甚至可以在用户对 Queue 内部实现一无所知的情况下写出。 然而,在 Queue 上使用 == 蕴涵了一个合理 Eq 的假设。 假设测试者写出了合适的生成器 gen_queue 并且完成了上面 6 个性质的测试, 这是否能保证 Queue 的实现是完全正确的呢?很遗憾这并不一定。 下面我们考虑如下构造:

///|
struct Queue {
  f : @list.List[Int]
  r : @list.List[Int]
} derive(Show)

///|
fn bq(f : @list.List[Int], r : @list.List[Int]) -> Queue {
  match f {
    Empty => { f: r.rev(), r: @list.empty() }
    _ => { f, r }
  }
}

///|
fn enqueue(x : Int, q : Queue) -> Queue {
  bq(q.f, q.r.prepend(x))
}

///|
fn empty() -> Queue {
  bq(@list.empty(), @list.empty())
}

///|
fn is_empty(q : Queue) -> Bool {
  q.f.is_empty()
}

///|
fn front(q : Queue) -> Int {
  q.f.unsafe_last()
}

///|
fn dequeue(q : Queue) -> Queue {
  let { f, r } = q
  bq(f.unsafe_tail(), r)
}

熟悉算法的读者可以注意到这里的 front 实现是错误的:它取了 f 的尾部而非头部。 接下来我们再给 == 给出定义:

///|
impl Eq for Queue with equal(self, other) {
  let to_list = (q : Queue) => q.f.concat(q.r.rev())
  to_list(self) == to_list(other)
}

///|
fn q1() -> Bool {
  is_empty(empty()) == true
}

///|
fn q2(xq : (Int, Queue)) -> Bool {
  let (x, q) = xq
  is_empty(enqueue(x, q)) == false
}

///|
fn q3(x : Int) -> Bool {
  front(enqueue(x, empty())) == x
}

///|
fn q4(xq : (Int, Queue)) -> Bool {
  let (x, q) = xq
  guard !is_empty(q) else { true }
  front(enqueue(x, q)) == front(q)
}

///|
fn q5(x : Int) -> Bool {
  dequeue(enqueue(x, empty())) == empty()
}

///|
fn q6(xq : (Int, Queue)) -> Bool {
  let (x, q) = xq
  guard !is_empty(q) else { true }
  dequeue(enqueue(x, q)) == enqueue(x, dequeue(q))
}

由于 == 被定义为「转换为列表再比较」,也就是一种「语义相等」, 所以 Q1 ~ Q6 公理性质会成立,可以使用下面代码验证之:

///|
fn gen_int_list() -> @qc.Gen[@list.List[Int]] {
  @qc.sized(fn(n) { @qc.list_with_size(n, @qc.small_int()) })
}

///|
fn gen_queue() -> @qc.Gen[Queue] {
  let gl = gen_int_list()
  gl.bind(fn(f) { gl.bind(fn(r) { @qc.pure(bq(f, r)) }) })
}

///|
test "queue axioms q1-q6" {
  let gen_xq = @qc.tuple(@qc.small_int(), gen_queue())
  let gen_x = @qc.small_int()
  @qc.quick_check(q1())
  @qc.quick_check(@qc.forall(gen_xq, q2))
  @qc.quick_check(@qc.forall(gen_x, q3))
  @qc.quick_check(@qc.forall(gen_xq, q4))
  @qc.quick_check(@qc.forall(gen_x, q5))
  @qc.quick_check(@qc.forall(gen_xq, q6))
}

上面的测试在随机意义上会「全部通过」,因为它们只约束 == 的等价类行为,而该等价恰好 与错误实现保持一致。我们可以在语义层面看出问题:对于 FIFO 队列,连续入队 1、2、3 后再出队, 前端元素应当是 2,但错误实现因为把 front 写成取尾部而返回 3。由此可见,仅依赖公理性质并不 等价于验证了「可观察行为」的正确性。

这是因为我们在用公理做等式推理时, 隐含假设了操作对相等是同余;而测试只测了公理,没有测这个隐含假设。 这一问题本质上是「可观察等价」与「程序等价」之间的错位。 用户只能通过公开操作观察行为,因此当两个 值在 == 意义下被视为等价时,我们期待任何观察操作都给出等价结果。 若这一点不成立,则实现对外行为已经偏离规范,即便所有基本公理仍然为真。

为弥补上述不足,我们引入「操作不变性」测试,即当两个值在 == 意义下等价时,任何观察操作的结果 也应当等价。直接用随机 q 与 q1 测试这一点往往会失败于「等价样本稀少」, 因此我们考虑等价对的生成器(@qc.Equivalence), 并在此基础上定义兼容性性质:

///|
fn from_list(xs : @list.List[Int]) -> @qc.Gen[Queue] {
  let len = xs.length()
  let gen_i = if len <= 0 { @qc.pure(0) } else { @qc.int_range(0, len + 1) }
  gen_i.fmap(fn(i) {
    let xs1 = xs.take(i)
    let xs2 = xs.drop(i)
    bq(xs1, xs2.rev())
  })
}

///|
fn gen_equiv_queue() -> @qc.Gen[@qc.Equivalence[Queue]] {
  gen_int_list().bind(fn(z) {
    from_list(z).bind(fn(x) { from_list(z).fmap(fn(y) { { lhs: x, rhs: y } }) })
  })
}

///|
test "queue invariance" {
  let prop = @qc.forall(gen_equiv_queue(), eqv => {
    let { lhs, rhs } = eqv
    guard !is_empty(lhs) else { true }
    front(lhs) == front(rhs)
  })
  @qc.quick_check(prop, expect=Fail)
}

这类测试能够更直接地暴露 front 的实现错误,但同时也显露出一个现实约束:等价对的生成依赖于对 内部表示的「扰动方式」,而这通常不由使用者掌握,换句话说,它依赖于对内部表示的深入理解与等价对的构造能力。 甚至如果等价对生成器写得过于保守,测试仍可能给出「全部通过」的假象。因此我们需要一种更系统、更自动化的生成机制。

为此,我们进一步提出第二条路径, 通过「单步公理重写」来系统派生测试。具体做法是把某条公理的左、右两侧分别代入 某个操作的某个参数位置,其余参数用随机值填充,并在必要时补充前置条件,从而获得可执行的操作不变性 性质集合。该方法牺牲了完备性,却显著降低了生成难度与测试成本。

这个方法的核心观察是,若要验证 == 对所有操作是同余的,原则上我们是想要验证对于每个操作 f 我们有:

t≡t′  ⟹  f(…,t,…)≡f(…,t′,…)t \equiv t' \implies f(\ldots, t, \ldots) \equiv f(\ldots, t', \ldots)

不过问题是我们很难生成足够好的等价对 t ≡ t'。我们可以注意到: 只要这个属性真的会失败,那么沿着从 t 到 t' 的重写序列, 总存在某一步「单次公理应用」已经导致外层操作结果不同 (否则全链条都相等,靠传递性会推出最终也相等,导致矛盾)。

所以我们最终的测试方案是:对于某个我们期望测试的操作 f 和一个参数位置 i,以及公理 lhs = rhs, 我们可以构造如下性质:

f(x1,⋯ ,lhs(y1,⋯ ,ym),⋯ ,xn)=f(x1,⋯ ,rhs(y1,⋯ ,ym),⋯ ,xn)f(x_1, \cdots, \text{lhs}(y_1,\cdots, y_m), \cdots, x_n) = f(x_1, \cdots, \text{rhs}(y_1,\cdots, y_m), \cdots, x_n)
  • yjy_j 是公理中出现的变量,我们可以简单生成它们的随机值
  • xkx_k 是操作 f 中其他参数位置的变量,可以用简单随机值填充,并且左右两边保持一致,因为我们只关心 lhs 和 rhs 的替换效果
///|
fn enqueue_1_q3(xq : (Int, Queue)) -> Bool {
  let (x, q) = xq
  let lhs = front(enqueue(x, empty()))
  let rhs = x
  enqueue(lhs, q) == enqueue(rhs, q)
}

///|
fn enqueue_1_q4(xqp : (Int, Queue, Queue)) -> Bool {
  let (x, q, p) = xqp
  guard !is_empty(q) else { true }
  let lhs = front(enqueue(x, q))
  let rhs = front(q)
  enqueue(lhs, p) == enqueue(rhs, p)
}

///|
fn front_1_q6(xq : (Int, Queue)) -> Bool {
  let (x, q) = xq
  guard !is_empty(q) else { true }
  let lhs = dequeue(enqueue(x, q))
  let rhs = enqueue(x, dequeue(q))
  front(lhs) == front(rhs)
}

///|
test "operation invariance tests" {
  let gen_xq = @qc.tuple(@qc.small_int(), gen_queue())
  let gen_xqp = @qc.triple(@qc.small_int(), gen_queue(), gen_queue())
  @qc.quick_check(@qc.forall(gen_xq, enqueue_1_q3))
  @qc.quick_check(@qc.forall(gen_xqp, enqueue_1_q4))
  @qc.quick_check(@qc.forall(gen_xq, front_1_q6), expect=Fail)
}

这些函数名称与约定一致,明确标识了「操作位置 + 公理编号」的组合,例如 enqueue_1_q3 表示把公理 Q3 的左右项分别代入 enqueue 的第一个参数位置;front_1_q6 则表示把 Q6 的左右项 代入 front 的唯一参数位置 (1)。我们可以将它们视为「系统自动生成测试」的目标形态: 每条测试都围绕一个局部等价替换展开,而不是试图枚举完整的等价对, 从而把难度从「生成等价输入」转化为「生成可用前置条件的随机参数」。 现在,这个测试能够成功捕获 front 实现中的错误。

我们可以看到操作不变性测试的核心价值在于:它并不要求测试者了解实现内部结构, 却能在「等价输入的观察一致性」这一层面提供更强的保障。该方法在理论上并不完备, 因为它只考虑单步公理重写而非任意深度的嵌套重写, 但在工程实践中,这种权衡显著降低了测试成本, 并在队列示例中成功暴露了 front 的实现缺陷。

模型对照​

那么模型对照(model-based testing)则更像是「从语义出发」的规格: 你先写一个更简单、更可信的参考模型(model/specification),再声 明真实实现(system under test, SUT)在可观察行为上与模型一致。 不要直接测试复杂系统本身,而是给它配一个可以被穷举/推理/解释的影子世界。 当 SUT 失败时,模型往往不仅告诉你错了,还告诉你具体的语义错误。

何时使用 Model ?​

  • SUT 具有复杂状态(缓存、连接池、并发队列、LRU、索引等),直接写等式规律很难。
  • 你有一个朴素但可信的版本(慢一点、用列表/Map 表示、甚至直接用规范描述)
    • 例如说我们有一个朴素的冒泡排序实现,可以用来测试更复杂的快速排序
    • 或者我们在搬运软件到 MoonBit 中时,可以用原有实现作为模型
    • 编译器设计也可以用「解释器」作为模型,测试「编译 + 运行」的结果

Set 模型​

集合(Set)是最常见的数据类型之一。它能自然表达「无序且不重复」的语义, 假设我们实现了一个简单的集合类型,现在我们想要分析它是否正确地实现了插入与删除操作。 可以考虑用列表作为模型,列表天然支持插入与删除,并且我们可以通过排序与去重来模拟集合的行为。

///|
struct ModelSet[T](@list.List[T])

///|
fn[T] ModelSet::empty() -> ModelSet[T] {
  ModelSet(@list.empty())
}

///|
fn[T : Eq] ModelSet::contains(self : ModelSet[T], x : T) -> Bool {
  self.0.contains(x)
}

///|
fn[T : Eq] ModelSet::insert(self : ModelSet[T], x : T) -> ModelSet[T] {
  guard not(self.contains(x)) else { self }
  self.0.prepend(x)
}

///|
fn[T : Eq] ModelSet::remove(self : ModelSet[T], x : T) -> ModelSet[T] {
  ModelSet(self.0.filter(fn(y) { y != x }))
}

接下来我们定义一组命令,表示对集合的操作序列:

///|
enum Cmd {
  Insert(Int)
  Remove(Int)
  Contains(Int)
} derive(Show, @coreqc.Arbitrary)

///|
struct Trace(@list.List[Bool]) derive(Eq) // 记录 contains 的结果

///|
pub fn run_model(cmds : @list.List[Cmd]) -> (ModelSet[Int], Trace) {
  let step = (str : (ModelSet[Int], Trace), cmd : Cmd) => {
    let (st, tr) = str
    match cmd {
      Insert(x) => (st.insert(x), tr)
      Remove(x) => (st.remove(x), tr)
      Contains(x) => (st, Trace(tr.0.prepend(st.contains(x))))
    }
  }
  @list.List::fold(cmds, init=(ModelSet::empty(), Trace(@list.empty())), step)
}

对应的 SUTSet 上会运行同样的命令序列,然后我们对比两者的最终状态与 Trace 结果是否一致:

///|
type SUTSet[T] = @sorted_set.SortedSet[T] // 假设这是我们要测试的复杂实现

///|
pub fn run_sut(cmds : @list.List[Cmd]) -> (SUTSet[Int], Trace) {
  let step = (str : (SUTSet[Int], Trace), cmd : Cmd) => {
    let (st, tr) = str
    match cmd {
      Insert(x) => (st.add(x), tr)
      Remove(x) => (st.remove(x), tr)
      Contains(x) => (st, Trace(tr.0.prepend(st.contains(x))))
    }
  }
  @list.List::fold(cmds, init=(SUTSet::new(), Trace(@list.empty())), step)
}

///|
test "model-based testing for Set" {
  let gen = @qc.list_with_size(20, @qc.Gen::spawn())
  let prop = @qc.forall(gen, fn(cmds) {
    let (model_set, model_trace) = run_model(cmds)
    let (sut_set, sut_trace) = run_sut(cmds)
    let model_set_arr = model_set.0.sort()
    let sut_set = @list.from_array(sut_set.to_array())
    model_trace == sut_trace && model_set_arr == sut_set
  })
  @qc.quick_check(prop)
}

注意到此处我们此处还考虑了 SortedSet 内部的排序行为,不仅通过 Trace 对比查询结果, 还对比了最终集合的内容是否一致且有序,这种双重验证确保了 SUT 在所有操作后都与模型的一致性。

上面仅提到了尤其简单的一种情况,实际中模型可以更复杂,在编程语言的设计与实现中我们也常见 一种「解释器 + 编译器」的组合测试模式。其中解释器由「抽象机」实现,直接执行源代码, 编译器则把源代码翻译为目标代码并运行。我们可以用解释器作为模型,验证编译器生成的代码在所有输入下与解释器行为一致。 不过在这种情况下,设计输入生成并非易事,感兴趣的读者可以考虑查阅 program synthesis 相关文献。

在MoonBit中实现IntMap

· 阅读需 8 分钟

键值对容器是现代编程语言必备的标准库成员之一,它应用广泛,所以其基本操作的性能非常重要。函数式语言的键值对容器实现大多基于某种平衡二叉搜索树,这样实现的键值对容器在查找和插入操作上表现优秀,但在需要合并两个键值对容器时表现不佳,命令式语言常用的哈希表也不擅长合并操作。

IntMap是一种为整数特化的不可变键值对容器,它只能以整数为键,通过牺牲一定的通用性,它实现了高效的合并/取交集操作。本文将从最朴素的二叉字典树开始,逐步改进到IntMap.

二叉字典树​

二叉字典树是一种使用每个键的二进制表示决定其位置的二叉树,键的二进制表示是一长串有限的0和1,那么假如当前位是0,就向左子树递归,当前位为1则向右子树递归.

///|
enum BinTrie[T] {
  Empty
  Leaf(T)
  Branch(left~ : BinTrie[T], right~ : BinTrie[T])
}

要在二叉字典树里查找某个键对应的值,只需要依次读取键的二进制位,根据其值选择向左或者向右移动,直到到达某个叶子节点

此处读取二进制位的顺序是从整数最小位到最高位

fn[T] BinTrie::lookup(self : BinTrie[T], key : UInt) -> T? {
  match self {
    Empty => None
    Leaf(value) => Some(value)
    Branch(left~, right~) =>
      if key % 2U == 0 {
        left.lookup(key / 2)
      } else {
        right.lookup(key / 2)
      }
  }
}

为了避免创建过多空树,我们不直接调用值构造子,而是使用branch方法

fn[T] BinTrie::br(left : BinTrie[T], right : BinTrie[T]) -> BinTrie[T] {
  match (left, right) {
    (Empty, Empty) => Empty
    _ => Branch(left~, right~)
  }
}

Patricia Tree​

Patricia Tree在二叉字典树的基础上保存了更多信息以加速查找,在每个分叉的地方,它都保留子树中所有键的公共前缀(虽然此处是从最低位开始计算,但我们仍然使用前缀这种说法),并用一个无符号整数标记当前的分支位(branching bit).这样一来,查找时需要经过的分支数量大大减少。

///|
enum PatriciaTree[T] {
  Empty
  Leaf(key~ : Int, value~ : T)
  Branch(
    prefix~ : UInt,
    mask~ : UInt,
    left~ : PatriciaTree[T],
    right~ : PatriciaTree[T]
  )
}

///|
fn[T] PatriciaTree::lookup(self : PatriciaTree[T], key : Int) -> T? {
  match self {
    Empty => None
    Leaf(key=k, value~) => if k == key { Some(value) } else { None }
    Branch(prefix~, mask~, left~, right~) =>
      if !match_prefix(key=key.reinterpret_as_uint(), prefix~, mask~) {
        None
      } else if zero(key.reinterpret_as_uint(), mask~) {
        left.lookup(key)
      } else {
        right.lookup(key)
      }
  }
}

///|
fn get_prefix(key : UInt, mask~ : UInt) -> UInt {
  key & (mask - 1U)
}

///|
fn match_prefix(key~ : UInt, prefix~ : UInt, mask~ : UInt) -> Bool {
  get_prefix(key, mask~) == prefix
}

///|
fn zero(k : UInt, mask~ : UInt) -> Bool {
  (k & mask) == 0
}

现在branch方法可以做更多优化, 保证Branch节点的子树不含Empty.

///|
fn[T] PatriciaTree::branch(
  prefix~ : UInt,
  mask~ : UInt,
  left~ : PatriciaTree[T],
  right~ : PatriciaTree[T],
) -> PatriciaTree[T] {
  match (left, right) {
    (Empty, right) => right
    (left, Empty) => left
    _ => Branch(prefix~, mask~, left~, right~)
  }
}

插入与合并​

既然类型定义已经确定,接下来要做的就是实现插入和合并操作。由于插入操作也可以看作将一个只有一个叶节点的树与原本的树合并,所以我们优先介绍合并操作的实现。

我们首先讨论可以走捷径的情况:假设我们有两个非空树t0和t1,它们的最长公共前缀分别为p0和p1且p0和p1互不包含, 那么不管t0和t1有多大,合并它们的成本都是一样的,因为只需要创建一个新的Branch节点。我们通过辅助函数join来实现。

生成掩码的函数gen_mask利用了整数二进制补码的一个特性来寻找最低的分支位。

假设输入x的二进制表示为

00100100000

那么,x.lnot()得到

11011011111

加一得到

11011100000

跟原来的x进行按位与后,得到:

00000100000
///|
fn[T] join(
  p0 : UInt,
  t0 : PatriciaTree[T],
  p1 : UInt,
  t1 : PatriciaTree[T],
) -> PatriciaTree[T] {
  let mask = gen_mask(p0, p1)
  if zero(p0, mask~) {
    PatriciaTree::Branch(prefix=get_prefix(p0, mask~), mask~, left=t0, right=t1)
  } else {
    PatriciaTree::Branch(prefix=get_prefix(p0, mask~), mask~, left=t1, right=t0)
  }
}

///|
fn gen_mask(p0 : UInt, p1 : UInt) -> UInt {
  fn lowest_bit(x : UInt) -> UInt {
    x & (x.reinterpret_as_int().neg().reinterpret_as_uint())
  }

  lowest_bit(p0 ^ p1)
}

万事俱备,现在可以开始编写insert_with函数了。对Empty和Leaf分支的处理都非常直接,而对于Branch, 在前缀互不包含时调用join,其他情况则根据分支位选择一个分支递归下去。

///|
fn[T] PatriciaTree::insert_with(
  self : PatriciaTree[T],
  k : Int,
  v : T,
  combine~ : (T, T) -> T,
) -> PatriciaTree[T] {
  fn go(tree : PatriciaTree[T]) -> PatriciaTree[T] {
    match tree {
      Empty => Leaf(key=k, value=v)
      Leaf(key~, value~) as tree =>
        if key == k {
          PatriciaTree::Leaf(key~, value=combine(v, value))
        } else {
          join(
            k.reinterpret_as_uint(),
            Leaf(key=k, value=v),
            key.reinterpret_as_uint(),
            tree,
          )
        }
      Branch(prefix~, mask~, left~, right~) as tree =>
        if match_prefix(key=k.reinterpret_as_uint(), prefix~, mask~) {
          if zero(k.reinterpret_as_uint(), mask~) {
            PatriciaTree::Branch(prefix~, mask~, left=go(left), right~)
          } else {
            PatriciaTree::Branch(prefix~, mask~, left~, right=go(right))
          }
        } else {
          join(k.reinterpret_as_uint(), Leaf(key=k, value=v), prefix, tree)
        }
    }
  }

  go(self)
}

合并操作基本遵循相同的逻辑,略有不同的是它还要考虑前缀与掩码完全相同的情况。

///|
fn[T] PatriciaTree::union_with(
  combine~ : (T, T) -> T,
  left : PatriciaTree[T],
  right : PatriciaTree[T],
) -> PatriciaTree[T] {
  fn go(left : PatriciaTree[T], right : PatriciaTree[T]) -> PatriciaTree[T] {
    match (left, right) {
      (Empty, t) | (t, Empty) => t
      (Leaf(key~, value~), t) => t.insert_with(key, value, combine~)
      (t, Leaf(key~, value~)) =>
        t.insert_with(key, value, combine=fn(x, y) { combine(y, x) })
      (
        Branch(prefix=p, mask=m, left=s0, right=s1) as s,
        Branch(prefix=q, mask=n, left=t0, right=t1) as t,
      ) =>
        if m == n && p == q {
          // The trees have the same prefix. Merge the subtrees
          PatriciaTree::Branch(
            prefix=p,
            mask=m,
            left=go(s0, t0),
            right=go(s1, t1),
          )
        } else if m < n && match_prefix(key=q, prefix=p, mask=m) {
          // q contains p. Merge t with a subtree of s
          if zero(q, mask=m) {
            Branch(prefix=p, mask=m, left=go(s0, t), right=s1)
          } else {
            Branch(prefix=p, mask=m, left=s0, right=go(s1, t))
          }
        } else if m > n && match_prefix(key=p, prefix=q, mask=n) {
          // p contains q. Merge s with a subtree of t.
          if zero(p, mask=n) {
            Branch(prefix=q, mask=n, left=go(s, t0), right=t1)
          } else {
            Branch(prefix=q, mask=n, left=t0, right=go(s, t1))
          }
        } else {
          join(p, s, q, t)
        }
    }
  }

  go(left, right)
}

Big-endian Patricia Tree​

Big-endian Patricia Tree在Patricia Tree的基础上将计算分支位的顺序改成了从最高位到最低位,

这样做有什么好处呢?

  • 更好的局部性。在Big-endian Patricia Tree中,大小相近的整数键会被放在邻近的地方。

  • 便于高效地按顺序遍历键,只要普通地实现前序/后序遍历即可。

  • 常见情况下合并速度更快。在实践中,intmap里的整数键一般是连续的,这种情况下Big-endian Patricia Tree会有更长的公共前缀,让合并操作更快。

  • 在Big-endian Patricia Tree中,如果把键看作无符号整数,右子树的每个键都大于当前节点的键(反过来,左子树是小于)。在编写查找函数时,只要使用无符号整数的比较就可判断接下来往哪个分支走,在大多数机器上这只需要一条指令即可完成,成本较低。

由于最终版本IntMap的实现与前文所述的Little Endian Patricia Tree相差不大,此处不再赘述,有需要的读者可以参考此仓库中的实现:https://github.com/moonbit-community/intmap

原实现中的一处错误​

虽然IntMap的实现思路相当简洁明了,但在编写具体的实现代码时还是可能犯一些非常隐蔽的错误,甚至原论文作者在编写IntMap的SML实现时也未能幸免,后来又被OCaml的Ptset/Ptmap模块继承,直到2018年发表的QuickChecking Patricia Trees一文中这个问题才被发现。

具体来说,由于SML和OCaml语言没有提供无符号整数类型,在这两种语言的实现中IntMap类型里的掩码都通过int存储,但在union_with函数中对掩码进行比较时,他们都忘记了应该使用无符号整数的比较。

在 MoonBit 中实现 Shunting Yard 算法

· 阅读需 12 分钟

什么是 Shunting Yard 算法?​

在编程语言或解释器的实现中,如何处理数学表达式一直是一个经典问题。我们希望能够像人一样理解“中缀表达式”(如 3 + 4 * 2),并正确考虑运算符优先级与括号。

1961 年,Edsger Dijkstra 提出了著名的 Shunting Yard 算法,它提供了一种机械化的方式来将中缀表达式转换为后缀表达式(RPN)或抽象语法树(AST)。算法的名字来源于铁路编组场:火车车厢通过在轨道之间来回调度实现排序,而在表达式处理中,我们通过两个栈来存储和调度操作数与操作符。想象一下你在脑子里计算 3 + 4 * 2 的过程:

  1. 你知道乘法优先级更高,所以要先算 4 * 2。
  2. 在此过程中,你会把前面的 3 和 + 临时“记在心里”。
  3. 等乘法结果出来,再把它和 3 相加。

Dijkstra 的洞见在于:这种人类计算时“临时记住某些东西再回来处理”的思维过程,其实可以用栈(stack)来模拟。就像铁路编组场会把火车车厢先临时停放在侧轨,再根据需要调度一样,算法通过把数字和运算符在不同的栈之间移动,来实现对运算顺序的控制。“Shunting Yard”(调度场算法)的名字正是来自这种铁路类比:

  • 火车车厢通过在轨道间移动来完成排序;
  • 数学表达式中的运算符和数字,也可以通过在栈之间移动来完成正确的排序与计算。

Dijkstra 把我们人类零散、混乱的计算过程,抽象成了一个清晰、机械化的流程,让计算机也能按照同样的逻辑来处理算式。

Shunting Yard 算法的基本流程​

Shunting Yard 算法通过维护两个栈来保证表达式按照正确的优先级和结合性进行解析:

  1. 初始化

    建立两个空栈:

    • 运算符栈(op_stack),用于临时存放尚未处理的运算符和括号;
    • 值栈(val_stack),用于存放操作数以及已经构造好的部分子表达式。
  2. 逐一扫描输入 token

    • 若 token 为数字或变量:直接压入 val_stack。

    • 若 token 为运算符:

      1. 检查 op_stack 栈顶元素。
      2. 当且仅当栈顶运算符的优先级高于当前运算符,或优先级相等且为左结合时,将该栈顶运算符弹出,并结合 val_stack 中的两个操作数,合成一个新的子表达式,再压回 val_stack。
      3. 重复此过程,直到不满足条件为止,然后将当前运算符压入 op_stack。
    • 若 token 为左括号:压入 op_stack,作为分界标记。

    • 若 token 为右括号:不断从 op_stack 弹出运算符,并用 val_stack 顶部的操作数组合成子表达式,直到遇到左括号为止;左括号本身丢弃,不会进入 val_stack。

  3. 清空运算符栈

    当所有 token 扫描完成后,若 op_stack 中仍有运算符,则依次弹出,并与 val_stack 中的操作数组合成更大的表达式,直到运算符栈为空。

  4. 结束条件

    最终,val_stack 中应只剩下一个元素,该元素即为完整的抽象语法树或后缀表达式。若栈内元素数量不为一,或存在未匹配的括号,则说明输入表达式存在错误。

演算示例​

接下来我们以解析 (1 + 2) * (3 - 4) ^ 2 为例,来展示在读入 token 的过程中,两个栈是如何变化的,从而更好地理解 Shunting Yard 算法:

步骤读入 token运算符栈(op_stack)值栈(val_stack)说明
1([(][]左括号压入运算符栈
21[(][1]数字压入值栈
3+[(, +][1]运算符压入运算符栈
42[(, +][1, 2]数字压入值栈
5)[][1 + 2]弹出直到左括号:1 与 2 结合为 1+2
6*[*][1 + 2]运算符压入运算符栈
7([*, (][1 + 2]左括号压入运算符栈
83[*, (][1 + 2, 3]数字压入值栈
9-[*, (, -][1 + 2, 3]运算符压入运算符栈
104[*, (, -][1 + 2, 3, 4]数字压入值栈
11)[*][1 + 2, 3 - 4]弹出直到左括号:3 与 4 结合为 3-4
12^[*, ^][1 + 2, 3 - 4]幂运算符压入栈(右结合,不会触发弹出)
132[*, ^][1 + 2, 3 - 4, 2]数字压入值栈
14输入结束[][(1 + 2) * (3 - 4) ^ 2]清空运算符栈:先弹出 ^,结合 3-4 与 2;再弹出 *,结合 1+2 与结果

在这个例子中,有以下几点值得注意:

  • 括号优先处理 在第一组括号 (1 + 2) 中,运算符 + 被延迟存放在运算符栈中,直到遇到右括号才与 1 和 2 结合。第二组括号 (3 - 4) 的处理过程完全相同。

  • 优先级的体现 当遇到 * 时,它被压入运算符栈。但随后遇到幂运算符 ^ 时,由于 ^ 的优先级高于 *,且是右结合,因此直接压栈,而不会触发 * 的弹出。

  • 结合性的作用 幂运算符 ^ 通常定义为右结合,这意味着表达式 a ^ b ^ c 会被解析为 a ^ (b ^ c)。在本例中,(3-4) ^ 2 保持这种结合方式,正确构建了子表达式。

  • 最终结果 在输入结束后,运算符栈依次被清空,最终形成完整的表达式:

(1 + 2) * ((3 - 4) ^ 2)

在 MoonBit 中实现 Shunting Yard 算法​

首先我们需要定义表达式和 token 的类型:

enum Expr {
  Literal(Int)
  BinExpr(String, Expr, Expr)
} derive(Show)

enum Token {
  Literal(Int)
  Op(String)
  LeftParen
  RightParen
} derive(Show)

我们可以利用 MoonBit 中的正则表达式匹配语法,快速的实现一个简单的 tokenizer:

pub fn tokenize(input : StringView) -> Array[Token] raise {
  let tokens = []
  for str = input {
    lexmatch str {
      "[0-9]+" as n, rest => {
        tokens.push(Token::Literal(@strconv.parse_int(n)))
        continue rest
      }
      "[\-+*/^]" as o, rest => {
        tokens.push(Token::Op(o.to_string()))
        continue rest
      }
      "\(", rest => {
        tokens.push(Token::LeftParen)
        continue rest
      }
      "\)", rest => {
        tokens.push(Token::RightParen)
        continue rest
      }
      "[ \n\r\t]+", rest => continue rest
      "$", _ => break
      _ => fail("Invalid input")
    }
  }
  tokens
}

tokenize 函数的作用是把输入字符串分割成一系列 token:

  • 匹配数字 [0-9]+,转换为 Token::Literal;
  • 匹配四则运算符和幂运算符 [-+*/^],转换为 Token::Op;
  • 匹配括号 ( 与 ),分别转换为 LeftParen 和 RightParen;
  • 匹配空格、换行等空白字符则直接跳过;
  • 如果遇到不符合规则的字符,则报错。 通过 lexmatch 和正则表达式,整个分词过程既简洁又高效。

接下来我们定义一个全局的操作符表,用于存储操作符的优先级和结合性:

priv enum Associativity {
  Left
  Right
}

priv struct OpInfo {
  precedence : Int
  associativity : Associativity
}

let op_table : Map[String, OpInfo] = {
  "+": { precedence: 10, associativity: Left },
  "-": { precedence: 10, associativity: Left },
  "*": { precedence: 20, associativity: Left },
  "/": { precedence: 20, associativity: Left },
  "^": { precedence: 30, associativity: Right },
}

这里通过 op_table 定义了常见运算符的优先级与结合性:

  • + 和 - 的优先级最低(10),是左结合;
  • * 和 / 的优先级更高(20),同样是左结合;

  • ^(幂运算)的优先级最高(30),但它是右结合。

接下来我们定义一个辅助函数,用于判断在遇到一个新的运算符时,是否需要先处理(弹出)栈顶的运算符:

fn should_pop(top_op_info~ : OpInfo, incoming_op_info~ : OpInfo) -> Bool {
  top_op_info.precedence > incoming_op_info.precedence ||
  (
    top_op_info.precedence == incoming_op_info.precedence &&
    top_op_info.associativity is Left
  )
}

should_pop 的逻辑是 Shunting Yard 算法的核心之一:

  • 如果栈顶运算符的优先级高于新运算符,则应当先处理栈顶运算符;
  • 如果两者优先级相等,并且栈顶运算符是左结合,同样应当先处理栈顶运算符;
  • 否则,保留栈顶运算符,把新运算符直接压入栈。

接下来我们实现表达式的解析函数:

pub fn parse_expr(tokens : Array[Token]) -> Expr {
  let op_stack : Array[String] = []
  let val_stack : Array[Expr] = []
  fn push_binary_expr(top_op) {
    let right = val_stack.pop().unwrap()
    let left = val_stack.pop().unwrap()
    val_stack.push(Expr::BinExpr(top_op, left, right))
  }

  for token in tokens {
    match token {
      Literal(n) => val_stack.push(Expr::Literal(n))
      Op(incoming_op) => {
        let incoming_op_info = op_table[incoming_op]
        while true {
          match op_stack.last() {
            None => break
            Some(top_op) =>
              if top_op != "(" &&
                should_pop(top_op_info=op_table[top_op], incoming_op_info~) {
                op_stack.pop() |> ignore
                push_binary_expr(top_op)
              } else {
                break
              }
          }
        }
        op_stack.push(incoming_op)
      }
      LeftParen => op_stack.push("(")
      RightParen =>
        while op_stack.pop() is Some(top_op) {
          if top_op != "(" {
            push_binary_expr(top_op)
          } else {
            break
          }
        }
    }
  }
  while op_stack.pop() is Some(top_op) {
    push_binary_expr(top_op)
  }
  val_stack.pop().unwrap()
}

parse_expr 是整个 Shunting Yard 算法的核心实现:

  1. 数据结构准备

    • op_stack 存储运算符和括号;
    • val_stack 存储操作数或部分构造好的子表达式;
    • 内部函数 push_binary_expr 封装了一个小步骤:从值栈中弹出两个操作数,结合运算符,生成一个新的 BinExpr 节点,并压回值栈。
  2. 遍历 token

    • 数字:直接压入 val_stack。
    • 运算符:不断检查 op_stack 栈顶的运算符,如果优先级更高或需要先计算,则弹出并构造子表达式;直到不满足条件时,把新运算符压入栈。
    • 左括号:压入 op_stack,用于分隔子表达式。
    • 右括号:持续弹出运算符,并结合值栈中的操作数形成子表达式,直到遇到匹配的左括号。
  3. 清空运算符栈

    遍历完成后,可能仍有运算符滞留在 op_stack 中,这时需要逐一弹出,并结合值栈中的操作数,直到运算符栈为空。

  4. 返回结果

    最终,值栈中应当只剩下一个元素,它就是完整的抽象语法树。若不是这种情况,说明输入表达式存在语法错误。

最后我们可以定义一个简单的 eval 函数,以便于进行测试:

pub fn eval(expr : Expr) -> Int {
  match expr {
    Literal(n) => n
    BinExpr(op, left, right) =>
      match op {
        "+" => eval(left) + eval(right)
        "-" => eval(left) - eval(right)
        "*" => eval(left) * eval(right)
        "/" => eval(left) / eval(right)
        "^" => {
          fn pow(base : Int, exp : Int) -> Int {
            if exp == 0 {
              1
            } else {
              base * pow(base, exp - 1)
            }
          }

          pow(eval(left), eval(right))
        }
        _ => abort("Invalid operator")
      }
  }
}

///|
pub fn parse_and_eval(input : String) -> Int raise {
  eval(parse_expr(tokenize(input)))
}

并通过一些简单的测试用例来验证我们的实现:

test "parse_and_eval" {
  inspect(parse_and_eval("1 + 2 * 3"), content="7")
  inspect(parse_and_eval("2 ^ 3 ^ 2"), content="512")
  inspect(parse_and_eval("(2 ^ 3) ^ 2"), content="64")
  inspect(parse_and_eval("(1 + 2) * 3"), content="9")
  inspect(parse_and_eval("10 - (3 + 2)"), content="5")
  inspect(parse_and_eval("2 * (3 + 4)"), content="14")
  inspect(parse_and_eval("(5 + 3) / 2"), content="4")
  inspect(parse_and_eval("10 / 2 - 1"), content="4")
  inspect(parse_and_eval("1 + 2 + 3"), content="6")
  inspect(parse_and_eval("10 - 5 - 2"), content="3")
  inspect(parse_and_eval("5"), content="5")
  inspect(parse_and_eval("(1 + 2) * (3 + 4)"), content="21")
  inspect(parse_and_eval("2 ^ (1 + 2)"), content="8")
  inspect(parse_and_eval("1 + 2 * 3 - 4 / 2 + 5"), content="10")
  inspect(parse_and_eval("((1 + 2) * 3) ^ 2 - 10"), content="71")
  inspect(parse_and_eval("100 / (2 * 5) + 3 * (4 - 1)"), content="19")
  inspect(parse_and_eval("2 ^ 2 * 3 + 1"), content="13")
  inspect(parse_and_eval("1 + 2 * 3 ^ 2 - 4 / 2"), content="17")
}

小结​

Shunting Yard 算法的核心思想在于使用两个栈来显式管理运算过程:

  • 值栈(val_stack) 用于保存数字和已经组合好的部分子表达式;
  • 运算符栈(op_stack) 用于保存尚未处理的运算符和括号。

通过定义运算符的优先级与结合性,并在扫描 token 的过程中不断比较和弹出栈顶运算符,Shunting Yard 能够保证表达式被按照正确的顺序组合成抽象语法树(AST)。 最终,当所有 token 被读取并且运算符栈清空后,值栈中剩下的就是完整的表达式树。

这种方法直观地模拟了我们手工计算时的思路:先把暂时不能计算的内容“记下来”,等条件合适时再取出处理。它的流程清晰、实现简洁,非常适合作为学习表达式解析的起点。

之前 MoonBit Pearl 中发布过一篇介绍 Pratt parsing 的文章,两者都是解决“如何正确解析表达式优先级和结合性”的经典方法,但它们的思路截然不同。 Shunting Yard 使用循环和显式的数据结构,通过运算符栈和值栈来管理尚未处理的符号和部分子表达式,整个过程像是手工操纵两个栈,逻辑清晰且容易跟踪。Pratt Parser 则基于递归下降,每个 token 定义在不同上下文下的解析方式,解析的推进依赖语言运行时的调用栈:每一次递归调用都相当于把尚未完成的状态压入栈中,返回时再继续组合。换句话说,Pratt Parser 将“栈”的存在隐藏在递归调用里,而 Shunting Yard 则把这种状态管理显式化,用循环和数据结构来直接模拟出来。因此,可以认为 Shunting Yard 是将 Pratt Parser 中隐含在调用栈里的机制转写为显式的栈操作。前者步骤机械化,适合快速实现固定的运算符解析;后者更灵活,尤其在处理前缀、后缀或自定义运算符时更为自然。

使用 MoonBit 和 Wassette 构建安全的 WebAssembly 工具

· 阅读需 9 分钟

欢迎来到 MoonBit 和 Wassette 的世界!本教程将带您一步步构建一个基于 WebAssembly 组件模型的安全工具。通过一个实用的天气查询应用示例,您将学习如何利用 MoonBit 的高效性和 wassette 的安全特性,创建功能强大的 AI 工具。

wassette 和 MCP 简介​

MCP(Model Completion Protocol)是 AI 模型与外部工具交互的协议。当 AI 需要执行特定任务(如网络访问或数据查询)时,会通过 MCP 调用相应工具。这种机制扩展了 AI 的能力,但也带来安全挑战。

wassette 是微软开发的一个基于 WebAssembly 组件模型的运行时,为 AI 系统提供安全执行外部工具的环境。它通过沙箱隔离和精确的权限控制,解决了 AI 工具可能带来的安全风险。

wassette 让工具运行在隔离环境中,权限受策略文件严格限制,接口通过 WIT(WebAssembly Interface Type)清晰定义。同时,也利用 WIT 接口来生成工具交互的数据格式。

总体流程​

在开始之前,让我们先了解一下整体流程:

让我们开始这段奇妙的旅程吧!

第1步:安装必要工具​

首先,我们需要安装三个工具(我们假设已经安装 MoonBit 工具链):

  • wasm-tools:WebAssembly 工具集,用于处理和操作 Wasm 文件
  • wit-deps:WebAssembly 接口类型依赖管理器
  • wit-bindgen:WebAssembly 接口类型绑定生成器,用于生成语言绑定
  • wassette:基于 Wasm 组件模型的运行时,用于执行我们的工具

其中,wasm-tools wit-deps wit-bindgen 可通过 cargo 安装(需安装 Rust):

cargo install wasm-tools
cargo install wit-deps-cli
cargo install wit-bindgen-cli

或从 GitHub Release 下载:

wassette 需从 GitHub Release 下载:

第2步:定义接口​

接口定义是整个工作流程的核心。我们使用 WebAssembly 接口类型 (WIT) 格式来定义组件的接口。

首先,创建项目目录和必要的子目录:

mkdir -p weather-app/wit
cd weather-app

创建 wit/deps.toml​

在 wit 目录下创建 deps.toml 文件,定义项目依赖:

cli = "https://github.com/WebAssembly/wasi-cli/archive/refs/tags/v0.2.7.tar.gz"
http = "https://github.com/WebAssembly/wasi-http/archive/refs/tags/v0.2.7.tar.gz"

这些依赖项指定了我们将使用的 WASI(WebAssembly 系统接口)组件:

  • cli:提供命令行接口功能。在这个例子中未使用。
  • http:提供 HTTP 客户端和服务器功能。在这个例子中使用客户端功能。

然后,运行 wit-deps update。这个命令会获取依赖,并在 wit/deps/ 目录下展开。

创建 wit/world.wit​

接下来,创建 wit/world.wit 文件来定义我们的组件接口。 WIT 是一种声明式接口描述语言,专为 WebAssembly 组件模型设计。它允许我们定义组件之间如何交互,而不需要关心具体的实现细节。 具体详情可以查看 组件模型 手册。

package peter-jerry-ye:weather@0.1.0;

world w {
  import wasi:http/outgoing-handler@0.2.7;
  export get-weather: func(city: string) -> result<string, string>;
}

这个 WIT 文件定义了:

  • 一个名为 peter-jerry-ye:weather 的包,版本为 0.1.0
  • 一个名为 w 的世界(world),它是组件的主要接口
  • 导入 WASI HTTP 的对外请求接口
  • 导出一个名为 get-weather 的函数,它接受一个城市名称字符串,返回一个结果(成功时为天气信息字符串,失败时为错误信息字符串)

第3步:生成代码​

现在我们已经定义了接口,下一步是生成相应的代码骨架。我们使用 wit-bindgen 工具来为 MoonBit 生成绑定代码:

# 确保您在项目根目录下
wit-bindgen moonbit --derive-eq --derive-show --derive-error wit

这个命令会读取 wit 目录中的文件,并生成相应的 MoonBit 代码。生成的文件将放在 gen 目录下。

注:当前生成版本存在部分警告,之后会进行修复。

生成的目录结构应该如下:

.
├── ffi/
├── gen/
│   ├── ffi.mbt
│   ├── moon.pkg.json
│   ├── world
│   │   └── w
│   │       ├── moon.pkg.json
│   │       └── stub.mbt
│   └── world_w_export.mbt
├── interface/
├── moon.mod.json
├── Tutorial.md
├── wit/
└── world/

这些生成的文件包含了:

  • 基础的 FFI(外部函数接口)代码(ffi/)
  • 生成的导入函数(world/ interface/
  • 导出函数的包装器(gen/)
  • 待实现的 stub.mbt 文件

第4步:修改生成的代码​

现在我们需要修改生成的存根文件,实现我们的天气查询功能。主要需要编辑的是 gen/world/w/stub.mbt 文件以及同目录下的 moon.pkg.json。在此之前,先让我们添加一下依赖,方便后续实现:

moon update
moon add moonbitlang/x
{
  "import": [
    "peter-jerry-ye/weather/interface/wasi/http/types",
    "peter-jerry-ye/weather/interface/wasi/http/outgoingHandler",
    "peter-jerry-ye/weather/interface/wasi/io/poll",
    "peter-jerry-ye/weather/interface/wasi/io/streams",
    "peter-jerry-ye/weather/interface/wasi/io/error",
    "moonbitlang/x/encoding"
  ]
}

让我们看一下生成的存根代码:

// Generated by `wit-bindgen` 0.44.0.

///|
pub fn get_weather(city : String) -> Result[String, String] {
  ... // 这里是我们需要实现的部分
}

现在,我们需要添加实现代码,使用 HTTP 客户端请求天气信息。编辑 gen/world/w/stub.mbt 文件,编辑如下:

///|
pub fn get_weather(city : String) -> Result[String, String] {
  (try? get_weather_(city)).map_err(_.to_string())
}

///| 利用 MoonBit 错误处理机制,简化实现
fn get_weather_(city : String) -> String raise {
  // 创建请求
  let request = @types.OutgoingRequest::outgoing_request(
    @types.Fields::fields(),
  )
  // 为了天气,我们访问 wttr.in 来获取
  if request.set_authority(Some("wttr.in")) is Err(_) {
    fail("Invalid Authority")
  }
  // 我们采用最简单的格式
  if request.set_path_with_query(Some("/\{city}?format=3")) is Err(_) {
    fail("Invalid path with query")
  }
  if request.set_method(Get) is Err(_) {
    fail("Invalid Method")
  }
  // 发出请求
  let future_response = @outgoingHandler.handle(request, None).unwrap_or_error()
  defer future_response.drop()
  // 在这里,我们采用同步实现,等待请求返回
  let pollable = future_response.subscribe()
  defer pollable.drop()
  pollable.block()
  // 在请求返回后,我们获取结果
  let response = future_response.get().unwrap().unwrap().unwrap_or_error()
  defer response.drop()
  let body = response.consume().unwrap()
  defer body.drop()
  let stream = body.stream().unwrap()
  defer stream.drop()
  // 将数据流解码为字符串
  let decoder = @encoding.decoder(UTF8)
  let builder = StringBuilder::new()
  loop stream.blocking_read(1024) {
    Ok(bytes) => {
      decoder.decode_to(
        bytes.unsafe_reinterpret_as_bytes()[:],
        builder,
        stream=true,
      )
      continue stream.blocking_read(1024)
    }
    // 如果流被关闭,则视为 EOF,正常结束
    Err(Closed) => decoder.decode_to("", builder, stream=false)
    // 如果出错,我们获取错误信息
    Err(LastOperationFailed(e)) => {
      defer e.drop()
      fail(e.to_debug_string())
    }
  }
  builder.to_string()
}

这段代码实现了以下功能:

  1. 创建一个 HTTP 请求,目标是 wttr.in 天气服务
  2. 设置请求路径,包含城市名称和格式参数
  3. 发送请求并等待响应
  4. 从响应中提取内容
  5. 解码内容并返回天气信息字符串

这段代码使用了 WASI HTTP 接口来发送请求,以同步 API 进行交互。其中,defer 关键字确保资源在使用后被正确释放。

第5步:构建项目​

现在我们已经实现了功能,下一步是构建项目。

# 编译 MoonBit 代码,生成核心 WebAssembly 模块
moon build --target wasm
# 嵌入 WIT 接口信息,指定字符串编码
wasm-tools component embed wit target/wasm/release/build/gen/gen.wasm -o core.wasm --encoding utf16
# 将核心 Wasm 模块转化为 Wasm 组件模块
wasm-tools component new core.wasm -o weather.wasm

构建成功后,会在项目根目录生成 weather.wasm 文件,这就是我们的 WebAssembly 组件。

之后,我们将它加载到 wassette 的路径中。当然,也可以选择通过对话,让 AI 来进行动态加载,不仅可以加载本地文件,也可以加载远程服务器上的文件。

wassette component load file://$(pwd)/weather.wasm

第6步(可选):配置安全策略​

wassette 会严格控制 WebAssembly 组件的权限,这是确保工具安全性的关键部分。这也是构建安全 MCP 工具的核心环节,通过细粒度的权限控制,我们可以确保工具只能执行预期的操作。

AI 可以在运行时通过调用默认的 wassette 的工具来进行赋权。我们可以预先执行这些命令。在我们的例子中,我们希望它能够访问 wttr.in 这个网站。因此,我们可以运行如下指令:

wassette permission grant network weather wttr.in

第7步:与 AI 交互​

最后,我们可以使用 wassette 运行我们的组件,并与 AI 交互。以 VSCode Copilot 为例,我们修改 .vscode/mcp.json,添加服务器:

{
  "servers": {
    "wassette": {
      // 假设 wassette 被添加至路径中
      // 否则请填写 wassette 可执行文件所在路径
      "command": "wassette",
      "args": [
        "serve",
        // 我们在这里禁用动态加载以及动态授权等功能
        "--disable-builtin-tools",
        "--stdio"
      ],
      "type": "stdio"
    }
  },
  "inputs": []
}

在刷新重启 wassette 后,我们便可以询问 AI 当前某个城市的天气。

当然,如果我们允许使用动态加载功能,我们也可以和 AI 这么说:

用 wassette,加载组件 ./weather.wasm(注意使用 file schema),并查询深圳的天气

于是,AI 便会先后调用 load-component 以及 get-weather 两个工具,获取天气,并且给出最后回答:

组件已成功加载,深圳的天气是:☀️ +30°C。

总结​

到这里,我们成功创建了一个基于 WebAssembly 组件模型的安全 MCP 工具,它可以:

  1. 通过定义清晰的接口
  2. 利用 MoonBit 的高效性
  3. 在 wassette 的安全沙箱中运行
  4. 与 AI 进行交互

Wassette 目前还只是 0.3.4 的版本,还缺少 MCP 的很多概念,如提示词、工作区、反向获取用户指令和 AI 生成能力等。但是它向我们展示了一个快速通过 Wasm 组件模型构建 MCP 的例子。

MoonBit 将会持续优化对于组件模型的能力,包括添加即将到来的 WASIp3 中异步的能力,并简化开发流程。敬请期待!

哈希表避坑指南

· 阅读需 16 分钟
Rynco Maekawa

本文介绍了哈希表的结构,演示了哈希表所面临的一个常见攻击手段——哈希洪泛攻击(hash flooding),以及如何在实践中消除这一攻击。

谁不喜欢哈希表呢?

它能以快如闪电的平均 O(1)O(1) 访问速度* 联系键值对, 而你只需要提供两样东西:一个比较相等的函数和一个哈希函数,就这么简单。 这一独特的性质使得哈希表在效率上常常优于其他关联性数据结构(如搜索树)。 因此,哈希表现已成为编程语言中使用最广泛的数据结构之一。

从 Python 中平平无奇的 dict,到数据库和分布式系统, 再到 JavaScript 对象,哈希表无处不在。 它们支撑着数据库的索引系统,实现了高效的缓存机制, 并构成了 Web 框架请求路由的骨干。 现代编译器用它们来管理符号表,操作系统靠它们来进行进程管理, 几乎每一个 Web 应用都用它们来维护用户状态。

无论你是在构建 Web 服务器、解析 JSON, 还是在处理配置文件,亦或只是统计词频, 你都很可能会用到哈希表。 它们已经变得如此基础,以至于许多开发者都将它们 O(1)O(1) 的魔法视为理所当然—— 但你有没有想过,这个 O(1)O(1) 的理所当然*到底是什么呢?

哈希表的内部构造​

一个哈希表由两部分组成: 一个桶数组和一个哈希函数。

struct MyHashMap[K, V] {
  buckets : Array[Bucket[K, V]]
  hash_fn : (K) -> UInt
}

桶数组包含了一系列所谓的"桶"。 每个桶都存储着我们插入的一些数据。

哈希函数 H 会为每个键(key)关联一个整数。 这个整数被用来在桶数组中寻找一个索引位置,以存储我们的值。 这个索引通常是通过将该整数与桶数组的大小进行取模运算得出的, 即 index = H(key) % bucket_array_size。 哈希表期望这个函数满足两个重要性质:

  1. 相同的键总是被转换成相同的数字。即,若 a == b,则 H(a) == H(b)。

    这个性质确保了, 我们用某个键存入数据后, 下次还能用同一个键准确地找到原来的位置。

  2. 对于不同的键,哈希函数产生的结果会尽可能均匀地分布在所有可能的结果空间中。

    这个性质确保了不同的键会大概率被转换到不同的整数值, 因此不太可能被映射到同一个桶中, 从而保证了检索的效率。

现在你可能会问,如果两个键被映射到了同一个桶,会发生什么呢? 那我们就不得不提哈希冲突了。

哈希冲突​

当两个键的哈希值相同时, 或者更广义地说,当两个键被映射到同一个桶时, 就发生了哈希冲突。

由于哈希表的一切决策都基于哈希值(或桶索引), 这两个键在哈希表看来就变得一模一样了—— 它们应该被放在同一个地方, 但它们本身又并不相等,所以不能互相覆盖。

哈希表的设计者们有几种策略来处理冲突, 这些策略大致可分为两类:

  • 链地址法将这些冲突的键放在同一个桶里。 现在,每个桶可能包含多个键的数据,而不仅仅是一个。 当查找一个冲突的键时, 需要遍历该桶中的所有键。

    struct ChainingBucket[K, V] {
      values : Array[(K, V)]
    }

    Java 的 HashMap 就是这种方法的一个著名例子。

  • 开放地址法仍然坚持每个桶只放一个键, 但当键发生冲突时,会使用一种独立的策略来选择另一个桶的索引。 当查找一个键时,会按照这种策略的顺序进行搜索, 直到可以确定没有更多可能的匹配项为止。

    struct OpenAddressBucket[K, V] {
      hash: Int
      key: K
      value: V
    }

    MoonBit 标准库中的 Map 就是这种方法的一个例子。

无论哪种情况,当哈希冲突发生时, 我们都别无选择,只能遍历我们找到的那个桶对应的所有键值对, 来确定我们正在寻找的键是否存在。

为了简单起见,我们以一个使用链地址法的哈希表为例。哈希表的实现看起来大概是这样的:

typealias ChainingBucket as Bucket

/// 搜索键存储的位置。
///
/// 返回 `(桶索引, 键在桶中的索引?, 完成的搜索次数)`
fn[K : Eq, V] MyHashMap::search(self : MyHashMap[K, V], key : K) -> (Int, Int?, Int) {
  let hash = (self.hash_fn)(key)
  let bucket = (hash % self.buckets.length().reinterpret_as_uint()).reinterpret_as_int()
  // 结果
  let mut found_index = None
  let mut n_searches = 0
  // 遍历桶中所有的键值对。
  for index, keyvalue in self.buckets[bucket].values {
    n_searches += 1
    if keyvalue.0 == key { // 检查键是否匹配。
      found_index = Some(index)
      break
    }
  }
  return (bucket, found_index, n_searches)
}

/// 插入一个新的键值对。
///
/// 返回完成的搜索次数。
fn[K : Eq, V] MyHashMap::insert(self : MyHashMap[K, V], key : K, value : V) -> Int {
  let (bucket, index, n_searches) = self.search(key)
  if index is Some(index) {
    self.buckets[bucket].values[index] = (key, value)
  } else {
    self.buckets[bucket].values.push((key, value))
  }
  n_searches
}

这就是 O(1)O(1) 访问魔法背后所附带的条件—— 如果我们运气不好,就必须遍历所有东西。 这使得哈希表在最坏情况下的复杂度变成了 O(n)O(n), 其中 nn 是哈希表中的键的数量。

制造一场冲突​

对于我们用于哈希表的大多数哈希函数来说,这种冲突的最坏情况是很罕见的。 这意味着我们通常不需要为最坏情况而烦恼, 并且在绝大多数时间里都能享受到 O(1)O(1) 的速度。

除非有人, 也许是某个心怀恶意的黑客, 故意把你推入最坏情况。

一般来说,哈希函数都是确定性的,而且运算速度很快。 所以,即使不去对函数本身进行高级的密码学分析, 我们仍然可以通过暴力破解找到很多会相互冲突的键。1

fn find_collision(
  bucket_count : Int,
  target_bucket : Int,
  n_collision_want : Int,
  hash_fn : (String) -> UInt,
) -> Array[String] {
  let result = []
  let bucket_count = bucket_count.reinterpret_as_uint()
  let target_bucket = target_bucket.reinterpret_as_uint()
  for i = 0; ; i = i + 1 {
    // 生成一些字符串键。
    let s = i.to_string(radix=36)
    // 计算哈希值
    let hash = hash_fn(s)
    let bucket_index = hash % bucket_count
    let bucket_index = if bucket_index < 0 {
      bucket_index + bucket_count
    } else {
      bucket_index
    }
    // 检查它是否与我们的目标桶冲突。
    if bucket_index == target_bucket {
      result.push(s)
      if result.length() >= n_collision_want {
        break
      }
    }
  }
  result
}

哈希洪泛攻击​

手握这些会冲突的键,(扮演恶意黑客的)我们现在就可以攻击哈希表, 持续利用其最坏情况下的复杂度。

考虑以下情况:你正在向同一个哈希表中插入键, 但每个键都被映射到同一个桶中。 每次插入时,哈希表都必须遍历桶中所有现有的键, 以确定新键是否已经存在。

第一次插入与 0 个键比较, 第二次与 1 个键比较,第三次与 2 个键比较, 被比较的键的数量随着每次插入线性增长。 对于 nn 次插入,被比较的键的总数是:

0+1+⋯+(n−1)=n(n−1)2=n2+n20 + 1 + \dots + (n - 1) = \frac{n(n - 1)}{2} = \frac{n^2 + n}{2}

这 nn 次插入操作总共需要 O(n2)O(n^2) 次比较才能完成2, 而平均情况下只需要 O(n)O(n) 次比较。 这个操作将比它本应花费的时间长得多。

这种攻击不仅限于插入操作。 每当一个被攻击的键被查找时, 都会比较相同数量的键, 因此每一个本应是 O(1)O(1) 的操作现在都变成了 O(n)O(n)。 这些原本耗时可以忽略不计的哈希表操作现在会变得极其缓慢, 使得攻击者比以前更容易耗尽程序的资源。

这就是我们所说的哈希洪泛攻击(hash flooding attack), 得名于它用冲突的键 “淹没” 了哈希表的同一个桶。

我们可以用我们之前写的哈希表实现来演示这一点:

/// 一个通过 Fowler-Noll-Vo 哈希函数实现的简单字符串哈希器。
/// https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function
fn string_fnv_hash(s : String) -> UInt {
  // 现实中应该直接在 s 背后的数组上工作,这里为了演示使用了 encode
  let s_bytes = @encoding/utf16.encode(s)
  let mut acc : UInt = 0x811c9dc5
  for b in s_bytes {
    acc = (acc ^ b.to_uint()) * 0x01000193
  }
  acc
}

fn test_attack(
  n_buckets : Int,
  keys : Array[String],
  hash_fn : (String) -> UInt,
) -> Int {
  let map = { buckets: Array::makei(n_buckets, _ => { values: [] }), hash_fn }
  let mut total_searches = 0
  for key in keys {
    total_searches += map.insert(key, 0)
  }
  total_searches
}

test {
  println("演示哈希洪泛攻击")
  let bucket_count = 2048
  let target_bucket_id = 42
  let n_collision_want = 1000

  //
  println("首先,尝试插入不冲突的键。")
  let non_colliding_keys = Array::makei(n_collision_want,
    i => (i * 37).to_string(radix=36))
  let n_compares_nc = test_attack(
    bucket_count, non_colliding_keys, string_fnv_hash,
  )
  println(
    "1000个不冲突键的总比较次数:\{n_compares_nc}",
  )
  println("")

  //
  println("现在,我们希望所有键都冲突到 #\{target_bucket_id} 号桶。")
  let colliding_keys = find_collision(
    bucket_count, target_bucket_id, n_collision_want, string_fnv_hash,
  )
  println("找到了 \{colliding_keys.length()} 个冲突的键。")
  let n_compares_c = test_attack(bucket_count, colliding_keys, string_fnv_hash)
  println(
    "1000个冲突键的总比较次数:\{n_compares_c}",
  )

  //
  let increase = n_compares_c.to_double() / n_compares_nc.to_double()
  println("比较次数增加了 \{increase} 倍")
}

上面代码的输出是:

演示哈希洪泛攻击
首先,尝试插入不冲突的键。
1000个不冲突键的总比较次数:347

现在,使用冲突的键...
找到了 1000 个冲突的键。
1000个冲突键的总比较次数:499500
比较次数增加了 1439.4812680115274 倍

……可以直接看到,现在的插入操作慢了大约 1000 倍!

在现实中,尽管哈希表中的桶数不像我们的例子那样是固定的, 但它们通常遵循一定的增长序列, 比如翻倍或遵循一个预定义的素数列表。 这种增长模式使得桶的数量非常容易预测。 因此,即使攻击者不知道确切的桶数,也能发起哈希洪泛攻击。

缓解哈希洪泛攻击​

哈希洪泛攻击之所以能奏效,是因为攻击者确切地知道哈希函数是如何工作的, 以及它是如何与键插入哈希表的位置相关联的。 如果我们改变其中任何一个,攻击就不再有效了。

带"种子"的哈希函数​

到目前为止,最简单的方法是防止攻击者确切地知道哈希算法是如何工作的。 这听起来可能不可能, 但哈希函数的性质实际上只需要在单个哈希表内部保持一致就行了!

在哈希表中,我们其实不需要一个可以在任何地方使用的、全局统一的"哈希值", 因为哈希表压根不在乎表以外洪水滔天,只要表本身保持一致就可以了。 所以,只要简单地在不同的哈希表之间切换哈希函数, 我们就能让攻击者无法预测其行为。

但你可能会说:“可现实世界中的哈希算法不是无限供应的啊!”

其实它可以是。 还记得我们说哈希函数需要将值尽可能均匀地分布在结果空间中吗? 这意味着,对于一个足够好的哈希函数, 输入的微小变化会导致输出的巨大变化(被称为雪崩效应)。 因此,为了给每个哈希表一个独一无二的哈希函数, 我们只需要在输入我们想要哈希的数据之前, 先给它 “喂” 一些该哈希表独有的数据。 这被称为哈希函数的“种子"(seed)。 这样,我们只要通过调整种子的值,就能获得无限供应的不同哈希函数了。

让我们用一个带种子的哈希函数和两个使用不同种子的哈希表来演示一下,哈希种子是如何解决这个问题的:

/// FNV 哈希的修改版,允许使用种子。
fn string_fnv_hash_seeded(seed : UInt) -> (String) -> UInt {
  let seed_bytes = seed.to_le_bytes()
  fn string_fnv_hash(s : String) -> UInt {
    let s_bytes = @encoding/utf16.encode(s)
    let mut acc : UInt = 0x811c9dc5
    // 混入种子字节。
    for b in seed_bytes {
      acc = (acc ^ b.to_uint()) * 0x01000193
    }
    // 哈希字符串字节。
    for b in s_bytes {
      acc = (acc ^ b.to_uint()) * 0x01000193
    }
    acc
  }

  string_fnv_hash
}

test {
  println("演示洪水攻击的缓解措施")
  let bucket_count = 2048
  let target_bucket_id = 42
  let n_collision_want = 1000

  // 第一个表使用种子 42。
  let seed1 : UInt = 42
  println("我们使用种子 \{seed1} 来寻找冲突")
  let hash_fn1 = string_fnv_hash_seeded(seed1)
  let colliding_keys = find_collision(
    bucket_count, target_bucket_id, n_collision_want, hash_fn1,
  )
  let n_compares_c = test_attack(bucket_count, colliding_keys, hash_fn1)
  println(
    "使用种子 \{seed1} 时,1000个冲突键的总比较次数:\{n_compares_c}",
  )
  println("")

  // 第二个表使用不同的种子。这次我们用 100
  let seed2 : UInt = 100
  println(
    "现在我们为第二个表使用不同的种子,这次是 \{seed2}",
  )
  let hash_fn2 = string_fnv_hash_seeded(seed2)
  let n_compares_nc = test_attack(bucket_count, colliding_keys, hash_fn2)
  println(
    "对于那些本应在种子 \{seed1} 下冲突的1000个键,现在的总比较次数:\{n_compares_nc}",
  )
}

上面程序的输出是:

演示洪水攻击的缓解措施
我们使用种子 42 来寻找冲突
使用种子 42 时,1000个冲突键的总比较次数:499500

现在我们为第二个表使用不同的种子,这次是 100
对于那些本应在种子 42 下冲突的1000个键,现在的总比较次数:6342

我们可以看到, 在第一个表中冲突的键,在第二个表中不再冲突了。3 因此,我们通过这个简单的技巧成功地缓解了哈希洪泛攻击。

至于那个让每个哈希表随机化的种子从哪里来…… 对于能够访问外部随机源的程序(比如 Linux 的 /dev/urandom), 使用它通常是最佳选择。 对于无法访问这类源的程序(比如在 WebAssembly 沙箱中), 在同一个进程中使用相同的种子、不同进程使用不同的种子也是一个方案(Python 就是这么做的)。 甚至于,一个每次请求种子时就自增的简单计数器或许也已经足够了—— 对于攻击者来说,猜测已经创建过多少个哈希表仍然是比较困难的一件事。

其他选择​

Java 使用了另一种解决方案, 当太多的值占据同一个桶时,它会退而求其次,使用一棵二叉搜索树(红黑树)存储它们。 是,这要求键除了可哈希之外,还必须是可比较的, 但现在它保证了 O(log⁡n)O(\log n) 的最坏情况复杂度, 这总比 O(n)O(n) 要好得多。

这为什么对我们很重要?​

由于哈希表在程序中无处不在, 在一个程序中找到一个你能控制其键的哈希表是极其容易的。 尤其是在 Web 程序中每, 请求头、Cookie、查询参数和 JSON 请求体都是键值对, 并且通常存储在哈希表中,这可能使它们容易受到哈希洪泛攻击。

一个对程序有足够了解(编程语言、框架等)的恶意攻击者, 可以尝试向 Web API 端点发送精心构造的请求负载。 这些请求需要更长的时间来处理, 所以如果一个常规的拒绝服务(DoS)攻击需要每秒 n 个请求才能使服务器瘫痪, 那么哈希洪泛攻击可能只需要小一个数量级的攻击次数就能达到相同的效果。 这使得它对攻击者来说效率高得多。 这种攻击被称为 哈希拒绝服务(HashDoS) 攻击。

幸运的是,通过在哈希表中引入一些哪怕是轻微的不可预测模式 (例如每个进程的随机性或带密钥的哈希), 我们就可以使这类攻击变得异常困难,以至于对攻击者不再可行。 此外,由于这种攻击高度依赖于对目标应用的语言、框架、架构和实现的了解, 构造一个攻击本身就已经相当困难了, 而现代的、配置良好的系统则更难被利用。

总结​

哈希表为我们提供了强大的、平均时间复杂度为常数的访问方式—— 然而,这个"常数"的成立,是建立在一些假设之上的, 而这些假设有时会被攻击者打破。 一次有针对性的哈希洪泛攻击会迫使许多键进入同一个桶, 将 O(1)O(1) 的操作变成 O(n)O(n), 能非常高效地耗尽系统资源。

好消息是,缓解措施既简单又实用: 为你的哈希表引入一些不可预测性, 当仅靠哈希还不够时使用旁路信息,或者当行为看起来不对劲时重新哈希。 有了这些,我们就可以让我们的哈希表既快速又安全。

Footnotes​

  1. 顺便提一下,这也类似于比特币挖矿的工作原理: 找到一个值添加到现有字符串中, 使得整个内容的哈希值(逐位倒过来之后)模除某个给定值之后的结果等于零。 ↩

  2. 甚至有一个 Tumblr 博客专门记录编程语言中意料之外的二次方复杂度, 叫做 Accidentally Quadratic。 你甚至可以在 这里 找到一个与哈希表相关的例子——这个例子几乎算是一次手动引入的哈希洪泛攻击了。 ↩

  3. 你可能会注意到,这个数字仍然比我们用随机生成的不冲突键得到的数字要高一些。 这可能与 FNV 哈希函数的设计并非追求最高质量的输出有关。 由于两个种子非常接近,结果可能仍然存在一些相似性。 使用一个更好的哈希函数(甚至是像 SipHash 这样的加密安全哈希函数) 会大大减少这种影响。 ↩

使用 MoonBit 开发一个 HTTP 文件服务器

· 阅读需 17 分钟

在这篇文章中,我将会介绍如何使用 MoonBit 的异步编程功能和 moonbitlang/async 库,编写一个简单的 HTTP 文件服务器。如果你之前接触过 Python 语言,那么你可能知道,Python 有一个非常方便的内建 HTTP 服务器模块。只需要运行 python -m http.server,就能在当期文件夹启动一个文件服务器,用于局域网文件共享等用途。 在这篇文章中,我们将用 MoonBit 实现一个类似功能的程序,并借此了解 MoonBit 的异步编程支持。我们还将额外支持一个 python -m http.server 没有的实用功能:把整个文件夹打包成 zip 文件下载。

异步编程简史​

异步编程,能让程序具有同时处理多项任务的能力。例如,对于一个文件服务器来说,可能会有多个用户同时访问这个服务器,而服务器需要同时服务所有用户,让它们的体验尽可能流畅、低延时。在典型的异步程序,例如服务器中,每项任务的大部分时间都花在等待 IO 上,实际的计算时间占比较低。因此,我们并不需要很多的计算资源,也能同时处理大量任务。而这其中的诀窍,就是频繁地在多个任务之间切换: 如果某项任务开始等待 IO,那么就不要继续处理它,而是马上切换到不需要等待的任务上。

过去,异步程序往往是通过多线程的方式实现的:每项任务对应一个操作系统的线程。 然而,操作系统线程需要占用较多资源,而且在线程之间切换开销较大。 因此,进入 21 世纪后,实现异步程序的主要方式变成了事件循环。 整个异步程序的形态是一个巨大的循环,每次循环中, 程序检查哪些 IO 操作已经完成,然后运行那些等待着这些已完成的 IO 操作的任务, 直到它们发起下一次 IO 请求,重新进入等待状态。 在这种编程范式中,任务间的切换发生在同一个用户态的线程里,因此开销极低。

然而,手写事件循环是一件非常痛苦的事情。 因为同一个任务的代码会被拆散到多次不同的循环中执行,程序的逻辑变得不连贯了。 因此,基于事件循环的程序非常难编写和调试。 幸运的是,就像大部分其他现代编程语言一样,MoonBit 提供了原生的异步编程支持。 用户可以像写同步程序一样写异步代码,MoonBit 会自动把异步代码切分成不同的部分。 而 moonbitlang/async 库则提供了事件循环和各种 IO 原语的实现,负责把异步代码运行起来。

MoonBit 中的异步编程​

在 MoonBit 中,可以用 async fn 语法来声明一个异步函数。 异步函数看上去和同步函数完全一样,只不过它们在运行时可能在中途被打断, 一段时间后才继续恢复运行,从而实现多个任务间的切换。 在异步函数中可以正常使用循环等控制流构造,MoonBit 编译器会自动将它们变成异步的样子。

和许多其他语言不同,在调用异步函数时,MoonBit 不需要用 await 之类的特殊语法标记, 编译器会自动推断出哪些函数调用是异步的。 不过,如果你使用带有 MoonBit 支持的 IDE 或文本编辑器查看代码, 就会看到异步函数调用被渲染成了斜体、可能抛出错误的函数调用带有下划线。 因此,阅读代码时,依然可以一眼就找到所有异步的函数调用。

对于异步程序来说,另一个必不可少的组件是事件循环、任务调度和各种 IO 原语的实现。 这一点在 MoonBit 中是通过 moonbitlang/async 库实现的。 moonbitlang/async 库中提供了网络IO、文件IO、进程创建等异步操作的支持, 以及一系列管理异步编程任务的 API。 接下来,我们将会在编写 HTTP 文件服务器的途中介绍 moonbitlang/async 的各种功能。

HTTP 服务器的骨架​

典型的 HTTP 服务器的结构是:

  • 服务器监听一个 TCP 端口,等待来自用户的连接请求
  • 接受来自用户的 TCP 连接后,服务器从 TCP 连接中读取用户的请求,处理用户的请求并将结果发回给用户

这里的每一项任务,都应该异步地进行: 在处理第一个用户的请求时,服务器仍应不断等待新的连接,并第一时间响应下一个用户的连接请求。 如果有多个用户同时连接到服务器,服务器应该同时处理所有用户的请求。 在这个过程中,所有可能耗费较多时间的操作,例如网络 IO 和文件 IO,都应该是异步的, 它们不应该阻塞程序、影响其他任务的处理。

在 moonbitlang/async 中,有一个辅助函数 @http.run_server, 能够绑我们自动完成上述工作,搭建一个 HTTP 服务器并运行它:

async fn server_main(path~ : String, port~ : Int) -> Unit {
  @http.run_server(@socket.Addr::parse("[::]:\{port}"), fn (conn, addr) {
    @pipe.stderr.write("received new connection from \{addr}\n")
    handle_connection(path, conn)
  })
}

server_main 接受两个参数,其中, path 是文件服务器工作的路径,port 是服务器监听的端口。 在 moonbitlang/async 中,一切异步代码都是可以取消的, 而异步代码被取消时会抛出错误,所以所有异步函数都会抛出错误。 因此,在 MoonBit 中,async fn 默认就会抛出错误,无需再显式标注 raise。

在 server_main 中,我们使用 @http.run_server 创建了一个 HTTP 服务器并运行它。 @http 是 moonbitlang/async 中提供 HTTP 解析等支持的包 moonbitlang/async/http 的别名, @http.run_server 的第一个参数是服务器要监听的地址。 这里我们提供的地址是 [::]:port, 这表示监听端口 port、接受来自任何网络接口的连接请求。 moonbitlang/async 有原生的 IPv4/IPv6 双栈支持,因此这里的服务器可以同时接受 IPv4 连接和 IPv6 连接。 @http.run_server 的第二个参数是一个回调函数,用于处理来自用户的连接。 回调函数会接受两个参数,第一个是来自用户的连接, 类型是 @http.ServerConnection,由 @http.run_server 自动获取并创建。 第二个参数是用户的网络地址。 这里,我们使用 handle_connection 函数来处理用户的请求,这个函数的实现将在稍后给出。 @http.run_server 会自动创建一个并行的任务,并在其中运行 handle_connection。 因此,服务器可以同时运行多份 handle_connection、处理多个连接。

处理用户来自用户的请求​

接下来,我们开始实现实际处理用户请求的 handle_connection 函数。 handle_connection 接受两个参数,base_path 是文件服务器处理的路径, 而 conn 是来自用户的连接。

async fn handle_connection(
  base_path : String,
  conn : @http.ServerConnection,
) -> Unit {
  for {
    let request = conn.read_request()
    conn.skip_request_body()
    guard request.meth is Get else {
      conn
      ..send_response(501, "Not Implemented")
      ..write("This request is not implemented")
      ..end_response()
    }
    let (path, download_zip) = match request.path {
      [ ..path, .."?download_zip" ] => (path.to_string(), true)
      path => (path, false)
    }
    if download_zip {
      serve_zip(conn, base_path + path)
    } else {
      let file = @fs.open(base_path + path, mode=ReadOnly) catch {
        _ => {
          conn
          ..send_response(404, "NotFound")
          ..write("File not found")
          ..end_response()
          continue
        }
      }
      defer file.close()
      if file.kind() is Directory {
        if download_zip {
        } else {
          serve_directory(conn, file.as_dir(), path~)
        }
      } else {
        server_file(conn, file, path~)
      }
    }
  }
}

在 handle_connection 中,程序通过一个大循环来不断从连接中读取用户请求并处理。 每次循环中,我们首先通过 conn.read_request() 读取一个来自用户的请求。 conn.read_request() 只会读取 HTTP 请求的头部,这是为了允许用户流式地读取较大的 body。 由于我们的文件服务器只处理 Get 请求,我们不需要请求的 body 中包含任何信息。 因此,我们通过 conn.skip_body() 跳过用户请求的 body,以保证下一个请求的内容可以被正确读取。

接下来,如果遇到不是 Get 的请求,guard 语句的 else 块会被执行, 此时,guard 语句后面的代码会被跳过,我们可以进入下一次循环、处理下一个请求。 在 else 块中,通过 conn.send_response(..) 向用户发送一个 “不支持该请求” 的回复。 conn.send_response(..) 会发送回复的头部,这之后,我们用 conn.write(..) 向连接写入回复的主体内容。 在写完所有内容后,我们需要用 conn.end_response() 来表明已经写完了回复的所有内容。

这里,我们希望实现一个 python -m http.server 中没有的实用功能: 以 zip 的形式下载整个文件夹。 如果用户请求的 URL 的形式是 /path/to/directory?download_zip, 我们就把 /path/to/directory 打包成 .zip 文件发送给用户。 这一功能是通过 serve_zip 函数来实现的。

由于我们实现的是一个文件服务器, 用户的 GET 请求中指定的路径会直接映射到 base_path 下对应的路径。 @fs 是 moonbitlang/async 中提供文件 IO 支持的包 moonbitlang/async/fs 的别名。 这里我们使用 @fs.open 打开对应的文件。 如果打开文件失败了,我们向用户发送一个 404 回复,告诉用户这个文件不存在。

如果用户请求的文件是存在的,那么我们需要把文件发送给用户。 当然,在此之前,别忘了用 defer file.close() 保证 file 占用的资源被及时释放。 通过 file.kind(),我们可以获得文件的种类。 在文件服务器中,如果用户请求的路径是一个文件夹,我们需要进行特殊的处理。 因为文件夹不能直接被发送给用户,我们需要根据文件夹的内容, 向用户返回一个 HTML 页面,让用户可以从页面看到文件夹里有哪些文件,并通过点击跳转到对应的页面。 这部分功能通过函数 serve_directory 提供。 如果用户请求的是一个普通文件,那么直接将文件的内容传输给用户即可。 这部分功能通过函数 serve_file 来实现。

向用户发送一个普通文件的代码如下:

async fn server_file(
  conn : @http.ServerConnection,
  file : @fs.File,
  path~ : String,
) -> Unit {
  let content_type = match path {
    [.., .. ".png"] => "image/png"
    [.., .. ".jpg"] | "jpeg" => "image/jpeg"
    [.., .. ".html"] => "text/html"
    [.., .. ".css"] => "text/css"
    [.., .. ".js"] => "text/javascript"
    [.., .. ".mp4"] => "video/mp4"
    [.., .. ".mpv"] => "video/mpv"
    [.., .. ".mpeg"] => "video/mpeg"
    [.., .. ".mkv"] => "video/x-matroska"
    _ => "appliaction/octet-stream"
  }
  conn
  ..send_response(200, "OK", extra_headers={ "Content-Type": content_type })
  ..write_reader(file)
  ..end_response()
}

这里,在 HTTP 回复中,我们根据文件的后缀名填入了不同的 Content-Type 字段。 这样一来,用户在浏览器中打开图片/视频/HTML 文件时,就可以直接预览文件的内容, 而不需要先下载文件再在本地打开。 对于其他文件,Content-Type 字段的值会是 application/octet-stream, 这会让浏览器自动将文件下载到本地。

我们依然使用 conn.send_response 来用户发送回复。 通过 extra_headers 字段我们可以在回复中加入额外的 HTTP header。 回复的主体则是文件的内容。 这里,conn.write_reader 会自动流式地把 file 的内容发送给用户。 假设用户请求了一个视频文件并在浏览器中播放, 如果我们先把整个视频文件读到内存中再发送给用户, 那么用户需要等服务器读入整个视频文件之后才能收到回复,服务器的响应速度会变慢。 而且,读入整个视频文件会浪费大量的内存。 而通过使用 write_reader,@http.ServerConnection 会自动把文件内容切成小块分段发送, 用户马上就能看到视频开始播放,占用的内存也会大大减少。

接下来,让我们实现显示文件夹的函数 serve_directory:

async fn serve_directory(
  conn : @http.ServerConnection,
  dir : @fs.Directory,
  path~ : String,
) -> Unit {
  let files = dir.read_all()
  files.sort()
  conn
  ..send_response(200, "OK", extra_headers={ "Content-Type": "text/html" })
  ..write("<!DOCTYPE html><html><head></head><body>")
  ..write("<h1>\{path}</h1>\n")
  ..write("<div style=\"margin: 1em; font-size: 15pt\">\n")
  ..write("<a href=\"\{path}?download_zip\">download as zip</a><br/><br/>\n")
  if path[:-1].rev_find("/") is Some(index) {
    let parent = if index == 0 { "/" } else { path[:index].to_string() }
    conn.write("<a href=\"\{parent}\">..</a><br/><br/>\n")
  }
  for file in files {
    let file_url = if path[path.length() - 1] != '/' {
      "\{path}/\{file}"
    } else {
      "\{path}\{file}"
    }
    conn.write("<a href=\"\{file_url}\">\{file}</a><br/>\n")
  }
  conn
  ..write("</div></body></html>")
  ..end_response()
}

这里,我们首先读入文件夹中的文件列表并对它们进行排序。 接下来,我们根据文件夹的内容,拼出一段 HTML 页面。 HTML 页面的主体内容是文件夹中的文件, 每个文件对应一个链接,上面显示着文件名,点击链接就能跳转到对应的文件。 这里,我们通过 HTML 的 <a> 元素来实现这一点。 如果文件夹不是根目录,那么我们在页面开头放上一个特殊的链接 ..,点击它会跳转到上一级目录。 此外,页面里还有一个 download as zip 的链接, 点击这个链接就能把当前文件夹打包成 zip 后下载。

实现将文件夹打包成 zip 的功能​

接下来,我们实现将文件夹打包成 zip 提供给用户的功能。 这里,简单起见,我们使用系统的 zip 命令。 serve_zip 函数的实现如下:

async fn serve_zip(
  conn : @http.ServerConnection,
  path : String,
) -> Unit {
  let full_path = @fs.realpath(path)
  let zip_name = if full_path[:].rev_find("/") is Some(i) {
    full_path[i+1:].to_string()
  } else {
    path
  }
  @async.with_task_group(fn(group) {
    let (we_read_from_zip, zip_write_to_us) = @process.read_from_process()
    defer we_read_from_zip.close()
    group.spawn_bg(fn() {
      let exit_code = @process.run(
        "zip",
        [ "-q", "-r", "-", path ],
        stdout=zip_write_to_us,
      )
      if exit_code != 0 {
        fail("zip failed with exit code \{exit_code}")
      }
    })
    conn
    ..send_response(200, "OK", extra_headers={
      "Content-Type": "application/octet-stream",
      "Content-Disposition": "filename=\{zip_name}.zip",
    })
    ..write_reader(we_read_from_zip)
    ..end_response()
  })
}

在 serve_zip 函数的开头,我们首先计算了用户下载的 .zip 文件的文件名。 接下来,我们使用 @async.with_task_group 创建了一个新的任务组。 任务组是 moonbitlang/async 中用于管理异步任务的核心构造, 所有异步任务都必须在一个任务组中创建。 在介绍 with_task_group 之前,让我们先看看 serve_zip 剩下的内容。 首先,我们使用 @process.read_from_process() 创建了一个临时管道, 从管道的一端写入的数据可以从另一侧读出,因此它可以用于读取一个进程的输出。 这里我们把管道的写入端 zip_write_to_us 会被提供给 zip 命令,用于写入压缩的结果。 而我们将从管道的读入端 we_read_from_zip 读取 zip 命令的输出,并将其发送给用户。

接下来,我们在新的任务组中创建了一个单独的任务, 并在其中使用 @process.run 运行 zip 命令。 @process 是 moonbitlang/async/process 的别名, 是 moonbitlang/async 中提供调用外部进程功能的包。 我们向 zip 传递的参数的意义是:

  • -q:不要输出日志信息
  • -r:递归压缩整个文件夹
  • -:把结果写入到 stdout
  • path:要压缩的文件夹

在调用 @process.run 时,我们通过 stdout=zip_write_to_us, 把 zip 命令的 stdout 重定向到了 zip_write_to_us,以获取 zip 的输出。 相比创建一个临时文件,这么做有两个好处:

  • 和 zip 间的数据传递完全在内存中进行,不需要进行低效的磁盘 IO
  • zip 一边压缩,我们可以一边像用户发送已经压缩好的部分,效率更高

@process.run 会等待 zip 结束运行,并返回 zip 命令的状态码。 如果 zip 的返回值不是 0,说明 zip 失败了,我们抛出一个错误。

在调用 zip 的同时,我们继续使用 conn.send_response(..) 向用户发送回复信息。 接下来,我们用 conn.write_reader(we_read_from_zip) 把 zip 的输出发送给用户。 Content-Disposition 这一 HTTP header 能让我们指定用户下载的 zip 文件的名字。

到这里,一切看上去都很合理。 但为什么这里要创建一个新的任务组呢?为什么不能直接提供创建新任务的 API 呢? 在编写异步程序时,有一个现象: 写出在正确时行为正确的程序比较容易,但写出在出错时依然行为正确的程序很难。 比如,对于 serve_zip 这个例子:

  • 如果 zip 命令失败了我们应该怎么办?
  • 如果数据发送到一半发生了网络错误,或者用户关闭了连接,应该怎么办?

如果 zip 命令失败了,那么整个 serve_zip 函数也应该失败。 由于此时用户可能已经收到了一部分不完整的数据,我们很难再把连接恢复到正常状态, 只能关闭把整个连接。 如果数据发送到一半发生了网络错误,那么我们应该停止 zip 的运行。 因为此时 zip 的结果已经没有用了,让它继续运行只是在浪费资源。 而且在最坏的情况下,由于我们不再读取 zip 的输出,和 zip 通信用的管道可能会被填满, 此时,zip 可能会永远阻塞在向管道写入的操作上,变成一个僵尸进程。

在上面的代码中,我们没有显式地写任何错误处理逻辑, 但是,在出现上述错误时,我们的程序的行为却是符合预期的, 而魔法就在于 @async.with_task_group 的语义,及其背后的 结构化并发 范式。 @async.with_task_group(f) 的大致语义如下:

  • 它会创建一个新的任务组 group,并运行 f(group)
  • f 可以通过 group.spawn_bg(..) 等函数在 group 中创建新的任务
  • 只有当 group 中的所有任务都完成时,with_task_group 才会返回
  • 如果 group 中的任何一个任务失败了,那么 with_task_group 也会失败,group 中的其他任务会被自动取消

这里的最后一条,就是保证正确错误处理的行为的关键:

  • 如果调用 zip 的任务失败了,那么错误会传播到整个任务组。 向用户发送回复的主任务会自动被取消, 然后错误会通过 with_task_group 自动向上传播,关闭连接
  • 如果发送回复的主任务失败了,错误同样会传播到整个任务组。 此时 @process.run 会被取消,此时它会自动向 zip 发送终止信号,结束 zip 的运行

因此,在使用 moonbitlang/async 编写异步程序时, 只需要根据程序的结构在适当的位置插入任务组, 剩下的错误处理的所有细节,都会由 with_task_group 自动解决。 这正是 moonbitlang/async 使用的结构化并发范式的威力:通过编程范式的引导, 它能让我们写出结构更清晰的异步程序,并以一种润物细无声的方式, 让异步程序在出错时也能有正确的行为。

让服务器跑起来​

至此,整个 HTTP 服务器的所有内容都已实现完毕,我们可以运行这个服务器了。 MoonBit 对异步代码有原生支持,可以直接用 async fn main 定义异步程序的入口, 或是用 async test 直接测试异步代码。 这里,我们让 HTTP 服务器运行在当前目录、向用户提供当前目录下的文件,并让它监听 8000 端口:

async test {
  server_main(path=".", port=8000)
}

通过 moon test moonbit_http_server.mbt.md 运行这份文档的源码, 并在浏览器中打开 http://127.0.0.1:8000,即可使用我们实现的文件服务器。

关于 moonbitlang/async 的更多功能,可以参考它的 API 文档 和 GitHub repo。

初探 MoonBit 中的 JavaScript 交互

· 阅读需 14 分钟


引言​

在当今的软件世界中,任何一门编程语言都无法成为一座孤岛。 对于 MoonBit 这样一门新兴的通用编程语言而言,若想在庞大的技术生态中茁壮成长,与现有生态系统的无缝集成便显得至关重要。

MoonBit 提供了包括 JavaScript 在内的多种编译后端,这为其对接广阔的 JavaScript 生态敞开了大门。 无论是对于浏览器前端开发,还是对于 Node.js 环境下的后端应用,这种集成能力都极大地拓展了 MoonBit 的应用场景,让开发者可以在享受 MoonBit 带来的类型安全与高性能的同时,复用数以万计的现有 JavaScript 库。

在本文中,我们将以 Node.js 环境为例,一步步探索 MoonBit JavaScript FFI 的奥秘,从基础的函数调用到复杂的类型与错误处理,向你展示如何优雅地搭建连接 MoonBit 与 JavaScript 世界的桥梁。

预先准备​

在正式启程之前,我们需要先为项目做好基础配置。如果还没有现成的项目,可以使用 moon new 工具创建一个新的 MoonBit 项目。

为了让 MoonBit 工具链知晓我们的目标平台是 JavaScript,我们需要在项目根目录的 moon.mod.json 文件中添加以下内容:

{
  "preferred-target": "js"
}

此项配置会告知编译器,在执行 moon build 或 moon check 等命令时,默认使用 JavaScript 后端。 当然,如果你希望在命令行中临时指定,也可以通过 --target=js 参数达到同样的效果。

编译项目​

完成上述配置后,只需在项目根目录下运行我们所熟悉的构建命令:

> moon build

命令执行成功后,由于我们的项目默认包含一个可执行入口,你可以在 target/js/debug/build/ 目录下找到编译产物。MoonBit 非常贴心地为我们生成了三个文件:

  • .js 文件:编译后的 JavaScript 源码。
  • .js.map 文件:用于调试的 Source Map 文件。
  • .d.ts 文件:TypeScript 类型声明文件,便于在 TypeScript 项目中集成。

第一个 JavaScript API 调用​

MoonBit 的 FFI 设计在原则上保持了一致性。与调用 C 或其他语言类似,我们通过一个带有 extern 关键字的函数声明来定义一个外部调用:

extern "js" fn consoleLog(msg : String) -> Unit = "(msg) => console.log(msg)"

这行代码是 FFI 的核心。让我们来分解一下:

  • extern "js":声明这是一个指向 JavaScript 环境的外部函数。

  • fn consoleLog(msg : String) -> Unit:这是该函数在 MoonBit 中的类型签名,它接受一个 String 类型的参数,并且返回一个单位值 (Unit)。

  • "(msg) => console.log(msg)":等号右侧的字符串字面量是这段 FFI 的“灵魂”,其中需要包含一段原生 JavaScript 函数。

    在这里,我们使用了一个简洁的箭头函数。 MoonBit 编译器会按原样将这段代码嵌入到最终生成的 .js 文件中,从而实现从 MoonBit 到 JavaScript 的调用。

    提示 如果你的 JavaScript 代码片段比较复杂,可以使用 #| 语法来定义多行字符串,以提高可读性。

一旦这个 FFI 声明就绪,我们就可以在 MoonBit 代码中像调用普通函数一样调用 consoleLog 了:

test "hello" {
  consoleLog("Hello from JavaScript!")
}

运行 moon test,你将会在控制台看到由 JavaScript console.log 打印出的信息。我们的第一座桥梁已经成功搭建!

JavaScript 类型的对接​

打通调用流程只是第一步,真正的挑战在于如何处理两种语言之间的类型差异。 MoonBit 是一门静态类型语言,而 JavaScript 则是动态类型语言。如何在这两者之间建立安全可靠的类型映射,是 FFI 设计中需要重点考虑的问题。

下面,我们从易到难,分情况介绍如何在 MoonBit 中对接不同的 JavaScript 类型。

无需转换的 JavaScript 类型​

最简单的情况是,MoonBit 中的某些类型在编译到 JavaScript 后端时,其底层实现本身就是对应的原生 JavaScript 类型。在这种情况下,我们可以直接进行传递,无需任何转换。

常见的“零成本”对接类型如下表所示:

MoonBit 类型JavaScript 对应类型
Stringstring
Boolboolean
Int, UInt, Float, Doublenumber
BigIntbigint
BytesUint8Array
Array[T]Array<T>
函数类型Function

基于这些对应关系,我们已经能够对许多简单的 JavaScript 函数进行绑定了。 事实上,在之前绑定 console.log 函数的例子中,我们已经使用了 MoonBit 中 String 类型与 JavaScript 中 string 类型的对应关系。

注意:维持 MoonBit 类型的内部不变量

一个非常重要的细节是,MoonBit 的所有标准数值类型(Int, Float 等)在 JavaScript 中都对应于 number 类型,即 IEEE 754 双精度浮点数。 这意味着当整数值越过 FFI 边界进入 JavaScript 后,其行为将遵循浮点数语义,这可能会导致在 MoonBit 看来非预期的结果,例如整数溢出行为的差异:

extern "js" fn incr(x : Int) -> Int = "(x) => x + 1"

test "incr" {
  // 在 MoonBit 中,@int.max_value + 1 会溢出并回绕
  inspect(@int.max_value + 1, content="-2147483648")
  // 在 JavaScript 中,它被当作浮点数处理,不会溢出
  inspect(incr(@int.max_value), content="2147483648") // ???
}

而这本质上是不合法的,因为根据 MoonBit 中 Int 的值的内部不变量,其值不可能是 2147483648(超出了类型允许的最大值)。 这可能导致下游依赖这一点的其他 MoonBit 代码出现意料之外的行为。 在跨越 FFI 边界处理其他数据类型时也有可能出现类似的问题,因此请在编写相关逻辑时务必留意这一点。

外部 JavaScript 类型​

当然,JavaScript 的世界远比上述基本类型要丰富。 我们很快就会遇到 undefined、null、symbol 以及各种复杂的宿主对象(Host Object)。这些类型在 MoonBit 中没有直接的对应物。

对于这种情况,MoonBit 提供了 #external 注解。 这个注解好比一个契约,它告诉编译器: “请相信我,这个类型在外部世界(JavaScript)中是真实存在的。 你不需要关心它的内部结构,只需把它当作一个不透明的句柄来处理即可。”

例如,我们可以这样定义一个代表 JavaScript undefined 的类型:

#external
type Undefined

extern "js" fn Undefined::new() -> Self = "() => undefined"

然而,单独的 Undefined 类型意义不大,因为在实际应用中,undefined 往往是作为联合类型(Union Type)的一部分出现的,例如 string | undefined。

一个更实用的方案是创建一个 Optional[T] 类型来精确对应 JavaScript 中的 T | undefined,并让它能与 MoonBit 内置的 T?(Option[T])类型方便地互相转换。

为了实现这个目标,我们首先需要一个能够代表“任意” JavaScript 值的类型,类似于 TypeScript 中的 any。这正是 #external 的用武之地:

#external
pub type Value

相应地,我们还需要提供获取 undefined 值和判断某值是否为 undefined 的方法:

extern "js" fn Value::undefined() -> Value =
  #| () => undefined

extern "js" fn Value::is_undefined(self : Self) -> Bool =
  #| (n) => Object.is(n, undefined)

为了方便调试,我们再为 Value 类型实现 Show 特质,让它可以被打印出来:

pub impl Show for Value with output(self, logger) {
  logger.write_string(self.to_string())
}

pub extern "js" fn Value::to_string(self : Value) -> String =
  #| (self) =>
  #|   self === undefined ? 'undefined'
  #|     : self === null ? 'null'
  #|     : self.toString()

接下来是整个转换过程中的“魔法”所在。我们定义两个特殊的转换函数:

fn[T] Value::cast_from(value : T) -> Value = "%identity"

fn[T] Value::cast(self : Self) -> T = "%identity"

何为 %identity

%identity 是 MoonBit 提供的一个特殊内建函数(intrinsic),它是一个“零成本”的类型转换操作。 它在编译时会进行类型检查,但在运行时不会产生任何效果。 它仅仅是告诉编译器:“作为开发者,我比你更清楚这个值的真实类型,请直接将它当作另一种类型来看待。”

这是一把双刃剑:它为 FFI 边界层的代码提供了强大的表达能力,但如果滥用,则可能破坏类型安全。 因此,它的使用场景应当被严格限制在 FFI 相关代码范围内。

有了这些积木,我们就可以开始搭建 Optional[T] 了:

#external
type Optional[_] // 对应 T | undefined

/// 创建一个 undefined 的 Optional
fn[T] Optional::undefined() -> Optional[T] {
  Value::undefined().cast()
}

/// 检查一个 Optional 是否为 undefined
fn[T] Optional::is_undefined(self : Optional[T]) -> Bool {
  self |> Value::cast_from |> Value::is_undefined
}

/// 从 Optional[T] 中解包出 T,如果为 undefined 则 panic
fn[T] Optional::unwrap(self : Self[T]) -> T {
  guard !self.is_undefined() else { abort("Cannot unwrap an undefined value") }
  Value::cast_from(self).cast()
}

/// 将 Optional[T] 转换为 MoonBit 内置的 T?
fn[T] Optional::to_option(self : Optional[T]) -> T? {
  guard !Value::cast_from(self).is_undefined() else { None }
  Some(Value::cast_from(self).cast())
}

/// 从 MoonBit 内置的 T? 创建 Optional[T]
fn[T] Optional::from_option(value : T?) -> Optional[T] {
  guard value is Some(v) else { Optional::undefined() }
  Value::cast_from(v).cast()
}

test "Optional from and to Option" {
  let optional = Optional::from_option(Some(3))
  inspect(optional.unwrap(), content="3")
  inspect(optional.is_undefined(), content="false")
  inspect(optional.to_option(), content="Some(3)")
  let optional : Optional[Int] = Optional::from_option(None)
  inspect(optional.is_undefined(), content="true")
  inspect(optional.to_option(), content="None")
}

通过这套组合拳,我们成功地在 MoonBit 的类型系统中为 T | undefined 找到了一个安全且人体工学良好的表达方式。 同样的方法也可以用于对接 null、symbol、RegExp 等其他 JavaScript 特有的类型。

处理 JavaScript 错误​

一个健壮的 FFI 层必须能够优雅地处理错误。 默认情况下,如果在 FFI 调用中,JavaScript 代码抛出了一个异常,这个异常并不会被 MoonBit 的 try-catch 机制捕获,而是会直接中断整个程序的执行:

// 这是一个会抛出异常的 FFI 调用
extern "js" fn boom_naive() -> Value raise = "(u) => undefined.toString()"

test "boom_naive" {
  // 这段代码会直接让测试进程崩溃,而不是通过 `try?` 返回一个 `Result`
  inspect(try? boom_naive()) // failed: TypeError: Cannot read properties of undefined (reading 'toString')
}

正确的做法是在 JavaScript 层用 try...catch 语句将调用包裹起来,然后找到一种办法将成功的结果或捕获到的错误传递回 MoonBit。 当然,我们可以直接在 extern "js" 声明的 JavaScript 代码中这么做,但也存在更可复用的解决办法:

首先,我们定义一个 Error_ 类型来封装来自 JavaScript 的错误:

suberror Error_ Value

pub impl Show for Error_ with output(self, logger) {
  logger.write_string("@js.Error: ")
  let Error_(inner) = self
  logger.write_object(inner)
}

接着,我们定义一个核心的 FFI 包装函数 Error_::wrap_ffi。 它的作用是在 JavaScript 领域执行一个操作(op),并根据成功与否,调用不同的回调函数(on_ok 或 on_error):

extern "js" fn Error_::wrap_ffi(
  op : () -> Value,
  on_ok : (Value) -> Unit,
  on_error : (Value) -> Unit,
) -> Unit =
  #| (op, on_ok, on_error) => { try { on_ok(op()); } catch (e) { on_error(e); } }

最后,我们利用这个 FFI 函数和 MoonBit 的闭包,就可以封装出一个符合 MoonBit 风格、返回 T raise Error_ 的 Error_::wrap 函数:

fn[T] Error_::wrap(
  op : () -> Value,
  map_ok~ : (Value) -> T = Value::cast,
) -> T raise Error_ {
  // 定义一个变量,用于在闭包内外传递结果
  let mut res : Result[Value, Error_] = Ok(Value::undefined())
  // 调用 FFI,传入两个闭包,它们会根据 JS 的执行结果修改 res 的值
  Error_::wrap_ffi(op, fn(v) { res = Ok(v) }, fn(e) { res = Err(Error_(e)) })
  // 检查 res 的值,并返回相应的结果或抛出错误
  match res {
    Ok(v) => map_ok(v)
    Err(e) => raise e
  }
}

现在,我们可以安全地调用之前那个会抛出异常的函数了,并且能以纯 MoonBit 代码来处理可能发生的错误:

extern "js" fn boom() -> Value = "(u) => undefined.toString()"

test "boom" {
  let result = try? Error_::wrap(boom)
  inspect(
    (result : Result[Value, Error_]),
    content="Err(@js.Error: TypeError: Cannot read properties of undefined (reading 'toString'))",
  )
}

对接外部 JavaScript API​

至此,我们已经掌握了处理类型和错误的关键技术,是时候将目光投向更广阔的天地了——整个 Node.js 和 NPM 生态系统。 而这一切的入口,就是对 require() 函数的绑定。

extern "js" fn require_ffi(path : String) -> Value = "(path) => require(path)"

/// 一个更方便的包装,支持链式获取属性,例如 require("a", keys=["b", "c"])
pub fn require(path : String, keys~ : Array[String] = []) -> Value {
  keys.fold(init=require_ffi(path), Value::get_with_string)
}

// ... 其中 Value::get_with_string 的定义如下:

fn[T] Value::get_with_string(self : Self, key : String) -> T {
  self.get_ffi(Value::cast_from(key)).cast()
}

extern "js" fn Value::get_ffi(self : Self, key : Self) -> Self = "(obj, key) => obj[key]"

有了这个 require 函数,我们就可以轻松加载 Node.js 的内置模块,例如 node:path 模块,并调用它的方法:

// 加载 node:path 模块的 basename 函数
let basename : (String) -> String = require("node:path", keys=["basename"]).cast()

test "require Node API" {
  inspect(basename("/foo/bar/baz/asdf/quux.html"), content="quux.html")
}

更令人兴奋的是,使用同样的方法,我们还能调用 NPM 上的海量第三方库。让我们以一个流行的统计学计算库 simple-statistics 为例。

首先,我们需要像在一个标准的 JavaScript 项目中那样,初始化 package.json 并安装依赖。这里我们使用 pnpm,你也可以换成 npm 或 yarn:

> pnpm init
> pnpm install simple-statistics

准备工作就绪后,我们就可以在 MoonBit 代码中直接 require 这个库,并获取其中的 standardDeviation 函数:

let standard_deviation : (Array[Double]) -> Double = require(
  "simple-statistics",
  keys=["standardDeviation"],
).cast()

现在,无论是 moon run 还是 moon test,MoonBit 都能正确地通过 Node.js 加载依赖并执行代码,返回我们期望的计算结果。

test "require external lib" {
  inspect(standard_deviation([2, 4, 4, 4, 5, 5, 7, 9]), content="2")
}

这无疑是激动人心的。仅仅通过几行 FFI 代码,我们就将 MoonBit 的类型安全世界与 NPM 庞大、成熟的生态系统连接在了一起。

结语​

通过本文的探索,我们初步了解了如何在 MoonBit 语言中与 JavaScript 进行交互,从最基础的类型对接到复杂的错误处理,再到外部库的轻松集成。 这些功能在 MoonBit 的静态类型系统与作为动态类型语言的 JavaScript 之间架起了一座桥梁,这体现了 MoonBit 作为现代编程语言在跨语言互操作性方面的思考。 它让开发者既能享受到 MoonBit 的类型安全与现代化的语言特性,又能无缝访问 JavaScript 的庞大生态,为 MoonBit 拓宽了不可估量的应用前景。

当然,能力越大,责任也越大:FFI 虽然强大,但在实际开发中仍需谨慎处理类型转换和错误边界,确保程序的健壮性。

对于希望利用 JavaScript 库来扩展 MoonBit 应用功能的开发者来说,掌握这些 FFI 技术将是一项至关重要的技能。 通过合理运用这些技术,我们可以构建出既具有 MoonBit 语言优势,又能充分利用 JavaScript 生态资源的高质量应用程序。

如果希望了解关于 MoonBit 在 JavaScript 互操作方面的探索进展的更多内容,欢迎关注基于 MoonBit 构建的 Web 应用前端 mooncakes.io 及其背后的界面库 rabbit-tea。

正则表达式引擎的两种实现方法:导数与 Thompson 虚拟机

· 阅读需 12 分钟

正则表达式引擎的实现方式多样,不同方法在性能、内存消耗和实现复杂度上各有权衡。本文将介绍两种数学上等价但实际表现迥异的正则匹配方法:Brzozowski 导数方法和 Thompson 虚拟机方法。

这两种方法都基于相同的抽象语法树表示,为直接的性能对比提供了统一的基础。其核心思想在于:这些看似不同的方法实际上是用不同的计算策略来解决同一个问题——一个依靠代数变换,另一个则通过程序执行。

约定与定义​

为了建立统一的基础,两种正则表达式引擎都采用相同的抽象语法树(AST)表示,用树形结构来描述正则表达式的基本构造:

enum Ast {
  Chr(Char)
  Seq(Ast, Ast)
  Rep(Ast, Int?)
  Opt(Ast)
} derive(Show, ToJson, Hash, Eq)

此外,我们还提供了智能构造函数来简化正则表达式的构建:

fn Ast::chr(chr : Char) -> Ast {
  Chr(chr)
}

fn Ast::seq(self : Ast, other : Ast) -> Ast {
  Seq(self, other)
}

fn Ast::rep(self : Ast, n? : Int) -> Ast {
  Rep(self, n)
}

fn Ast::opt(self : Ast) -> Ast {
  @fs.
  Opt(self)
}

AST 定义了四种基本的正则表达式操作:

  1. Chr(Char) - 匹配单个字符字面量
  2. Seq(Ast, Ast) - 序列匹配,即一个模式紧跟另一个模式
  3. Rep(Ast, Int?) - 重复匹配,None 表示无限次重复,Some(n) 表示恰好重复 n 次
  4. Opt(Ast) - 可选匹配,相当于标准正则语法中的 pattern?

举个例子,正则表达式 (ab*)? 表示一个可选的序列('a' 后跟零个或多个 'b'),可以这样构建:

Ast::chr('a').seq(Ast::chr('b').rep()).opt()

Brzozowski 导数方法​

导数方法基于形式语言理论,通过代数变换来处理正则表达式。对于输入的每个字符,该方法计算正则表达式的"导数",实质上是在问:"消费掉这个字符后,还剩下什么需要匹配?"这样就得到了一个新的正则表达式,代表剩余的匹配模式。

为了明确表示导数和可空性,我们对基本的 Ast 类型进行了扩展:

enum Exp {
  Nil
  Eps
  Chr(Char)
  Alt(Exp, Exp)
  Seq(Exp, Exp)
  Rep(Exp)
} derive(Show, Hash, Eq, Compare, ToJson)

Exp 中各构造器的含义如下:

  1. Nil - 表示不可能匹配的模式,即空集
  2. Eps - 匹配空字符串
  3. Chr(Char) - 匹配单个字符
  4. Alt(Exp, Exp) - 表示选择(或),在多个模式间进行选择
  5. Seq(Exp, Exp) - 表示连接,将两个模式依次连接
  6. Rep(Exp) - 表示重复,对模式进行零次或多次重复

通过 Exp::of_ast 函数,我们可以将 Ast 转换为表达能力更强的 Exp 格式:

fn Exp::of_ast(ast : Ast) -> Exp {
  match ast {
    Chr(c) => Chr(c)
    Seq(a, b) => Seq(Exp::of_ast(a), Exp::of_ast(b))
    Rep(a, None) => Rep(Exp::of_ast(a))
    Rep(a, Some(n)) => {
      let sec = Exp::of_ast(a)
      let mut exp = sec
      for _ in 1..<n {
        exp = Seq(exp, sec)
      }
      exp
    }
    Opt(a) => Alt(Exp::of_ast(a), Eps)
  }
}

同样,我们也为 Exp 提供了智能构造函数来简化模式构建:

fn Exp::seq(a : Exp, b : Exp) -> Exp {
  match (a, b) {
    (Nil, _) | (_, Nil) => Nil
    (Eps, b) => b
    (a, Eps) => a
    (a, b) => Seq(a, b)
  }
}

不过,Alt 的智能构造函数特别重要——它保证构造出的 Exp 符合 Brzozowski 原论文中的"相似性"标准化要求。两个正则表达式如果能通过以下规则相互转换,就被认为是相似的:

A∣∅→AA∣B→B∣AA∣(B∣C)→(A∣B)∣C \begin{align} & A \mid \emptyset &&\rightarrow A \\ & A \mid B &&\rightarrow B \mid A \\ & A \mid (B \mid C) &&\rightarrow (A \mid B) \mid C \end{align}

因此,我们对 Alt 构造进行标准化,确保始终使用一致的结合律和选择顺序:

fn Exp::alt(a : Exp, b : Exp) -> Exp {
  match (a, b) {
    (Nil, b) => b
    (a, Nil) => a
    (Alt(a, b), c) => a.alt(b.alt(c))
    (a, b) => {
      if a == b {
        a
      } else if a > b {
        Alt(b, a)
      } else {
        Alt(a, b)
      }
    }
  }
}

nullable 函数用于判断一个模式是否能够在不消费任何输入的情况下成功匹配(即匹配空字符串):

fn Exp::nullable(self : Exp) -> Bool {
  match self {
    Nil => false
    Eps => true
    Chr(_) => false
    Alt(l, r) => l.nullable() || r.nullable()
    Seq(l, r) => l.nullable() && r.nullable()
    Rep(_) => true
  }
}

deriv 函数计算模式对于特定字符的导数,按照 Brzozowski 导数理论中定义的规则对模式进行变换。我们对规则进行了重新排列,使其与 deriv 函数的实现顺序保持一致:

Da∅=∅Daϵ=∅Daa=ϵDab=∅ for (a≠b)Da(P∣Q)=(DaP)∣(DaQ)Da(P⋅Q)=(DaP⋅Q)∣(ν(P)⋅DaQ)Da(P∗)=DaP⋅P∗ \begin{align} D_{a} \emptyset &= \emptyset \\ D_{a} \epsilon &= \emptyset \\ D_{a} a &= \epsilon \\ D_{a} b &= \emptyset & \text{ for }(a \neq b) \\ D_{a} (P \mid Q) &= (D_{a} P) \mid (D_{a} Q) \\ D_{a} (P \cdot Q) &= (D_{a} P \cdot Q) \mid (\nu(P) \cdot D_{a} Q) \\ D_{a} (P\ast) &= D_{a} P \cdot P\ast \\ \end{align}
fn Exp::deriv(self : Exp, c : Char) -> Exp {
  match self {
    Nil => self
    Eps => Nil
    Chr(d) if d == c => Eps
    Chr(_) => Nil
    Alt(l, r) => l.deriv(c).alt(r.deriv(c))
    Seq(l, r) => {
      let dl = l.deriv(c)
      if l.nullable() {
        dl.seq(r).alt(r.deriv(c))
      } else {
        dl.seq(r)
      }
    }
    Rep(e) => e.deriv(c).seq(self)
  }
}

为了简化实现,我们这里只进行严格匹配,也就是说模式必须匹配整个输入字符串。因此,只有在处理完所有输入字符后,我们才检查最终模式的可空性:

fn Exp::matches(self : Exp, s : String) -> Bool {
  loop (self, s.view()) {
    (Nil, _) => {
      return false
    }
    (e, []) => {
      return e.nullable()
    }
    (e, [c, .. s]) => {
      continue (e.deriv(c), s)
    }
  }
}

虚拟机方法​

虚拟机方法将正则表达式编译成简单虚拟机的字节码指令。这种方法把模式匹配问题转化为程序执行过程,虚拟机同时模拟非确定性有限自动机中所有可能的执行路径。

Ken Thompson 在 1968 年的经典论文中描述了一种将正则模式编译为 IBM 7094 机器代码的引擎。其关键思路是:通过维护多个执行线程来避免指数级回溯,这些线程同步地在输入中前进,每次处理一个字符,同时探索所有可能的匹配路径。

指令集与程序表示​

该虚拟机基于四种基本指令运行,它们分别对应 NFA 的不同操作:

enum Ops {
  Done
  Char(Char)
  Jump(Int)
  Fork(Int)
} derive(Show, ToJson)

每条指令在 NFA 模拟中都有其特定作用:Done 标记匹配成功完成,对应 Thompson 原设计中的 match;Char(c) 消费输入字符 c 并跳转到下一条指令;Jump(addr) 无条件跳转至地址 addr,即 Thompson 的 jmp;Fork(addr) 创建两条执行路径——一条继续执行下一条指令,另一条跳转到 addr,对应 Thompson 的 split。

Fork 指令是处理模式非确定性的关键,比如选择和重复操作,这些情况下需要同时探索多条执行路径。这直接对应了 NFA 中的 ε-转换,即执行流可以在不消费输入的情况下发生分支。

我们定义了 Prg 类型,它封装了指令数组并提供便捷的方法来构建和操作字节码程序:

type Prg Array[Ops] derive(Show, ToJson)

fn Prg::push(self : Prg, inst : Ops) -> Unit {
  self.inner().push(inst)
}

fn Prg::length(self : Prg) -> Int {
  self.inner().length()
}

fn Prg::op_set(self : Prg, index : Int, inst : Ops) -> Unit {
  self.inner()[index] = inst
}

AST 到字节码的编译​

Prg::of_ast 函数采用标准的 NFA 构造技术,将 AST 模式转换为虚拟机指令:

  1. Seq(a, b):

    code for a
    code for b
  2. Rep(a, None) (无界重复):

        Fork L1, L2
    L1: code for a
        Jump L1
    L2:
  3. Rep(a, Some(n)) (固定重复):

    code for a
    code for a
    ... (n times) ...
  4. Opt(a) (可选):

        Fork L1, L2
    L1: code for a
    L2:

需要注意的是,Fork 构造器只接受一个地址参数,这是因为我们总是希望在 Fork 指令后继续执行下一条指令。

fn Prg::of_ast(ast : Ast) -> Prg {
  fn compile(prog : Prg, ast : Ast) -> Unit {
    match ast {
      Chr(chr) => prog.push(Char(chr))
      Seq(l, r) => {
        compile(prog, l)
        compile(prog, r)
      }
      Rep(e, None) => {
        let fork = prog.length()
        prog.push(Fork(0))
        compile(prog, e)
        prog.push(Jump(fork))
        prog[fork] = Fork(prog.length())
      }
      Rep(e, Some(n)) =>
        for _ in 0..<n {
          compile(prog, e)
        }
      Opt(e) => {
        let fork_inst = prog.length()
        prog.push(Fork(0))
        compile(prog, e)
        prog[fork_inst] = Fork(prog.length())
      }
    }
  }

  let prog : Prg = []
  compile(prog, ast)
  prog.push(Done)
  prog
}

虚拟机执行循环​

在 Rob Pike 的实现中,虚拟机会在输入字符串结束后再执行一轮来处理最终的接受状态。为了明确这个过程,我们的 matches 函数采用两阶段方法来实现核心的虚拟机执行循环:

阶段一:字符处理。对于每个输入字符,处理当前上下文中所有活跃的线程。如果 Char 指令匹配当前字符,就在下一个上下文中创建新线程。Jump 和 Fork 指令会立即在当前上下文中产生新线程。处理完所有线程后,交换上下文并继续处理下一个字符。

阶段二:最终接受判断。处理完所有输入后,检查剩余线程中是否有 Done 指令。同时处理那些不消费输入的 Jump/Fork 指令。如果有任何线程到达 Done 指令,就返回 true。

fn Prg::matches(self : Prg, data : @string.View) -> Bool {
  let Prg(prog) = self
  let mut curr = Ctx::new(prog.length())
  let mut next = Ctx::new(prog.length())
  curr.add(0)
  for c in data {
    while curr.pop() is Some(pc) {
      match prog[pc] {
        Done => ()
        Char(char) if char == c => {
          next.add(pc + 1)
        }
        Jump(jump) =>
          curr.add(jump)
        Fork(fork) => {
          curr.add(fork)
          curr.add(pc + 1)
        }
        _ => ()
      }
    }
    let temp = curr
    curr = next
    next = temp
    next.reset()
  }
  while curr.pop() is Some(pc) {
    match prog[pc] {
      Done => return true
      Jump(x) => curr.add(x)
      Fork(x) => {
        curr.add(x)
        curr.add(pc + 1)
      }
      _ => ()
    }
  }
  false
}

在 Rob Pike 的原始博客中,他使用递归函数来处理 Fork 和 Jump 指令,以保证线程按优先级执行。而我们这里采用了类似栈的结构来管理所有执行线程,这样可以自然地维护线程优先级:

struct Ctx {
  deque : @deque.T[Int]
  visit : FixedArray[Bool]
}

fn Ctx::new(length : Int) -> Ctx {
  { deque: @deque.new(), visit: FixedArray::make(length, false) }
}

fn Ctx::add(self : Ctx, pc : Int) -> Unit {
  if !self.visit[pc] {
    self.deque.push_back(pc)
    self.visit[pc] = true
  }
}

fn Ctx::pop(self : Ctx) -> Int? {
  match self.deque.pop_back() {
    Some(pc) => {
      self.visit[pc] = false
      Some(pc)
    }
    None => None
  }
}

fn Ctx::reset(self : Ctx) -> Unit {
  self.deque.clear()
  self.visit.fill(false)
}

visit 数组用于过滤掉低优先级的重复线程。添加新线程时,我们先通过 visit 数组检查该线程是否已存在于 deque 中。如果已存在就直接丢弃;否则加入 deque 并标记为已访问。这个机制对于处理像 (a?)* 这样可能无限扩展的模式很重要,能够有效避免无限循环或指数级的线程爆炸。

基准测试与性能分析​

我们通过一个对很多正则表达式实现都构成挑战的病理性案例来比较这两种方法:

test (b : @bench.T) {
  let n = 15
  let txt = "a".repeat(n)
  let chr = Ast::chr('a')
  let ast : Ast = chr.opt().rep(n~).seq(chr.rep(n~))
  let exp = Exp::of_ast(ast)
  b.bench(name="derive", () => exp.matches(txt) |> ignore())
  let tvm = Prg::of_ast(ast)
  b.bench(name="thompson", () => tvm.matches(txt) |> ignore())
}

模式 (a?){n}a{n} 是回溯引擎中典型的指数爆炸案例。这个模式有 n 种不同的方式来匹配 n 个 'a' 字符,在朴素的实现中会产生指数级的搜索空间。

name     time (mean ± σ)         range (min … max)
derive     41.78 µs ±   0.14 µs    41.61 µs …  42.13 µs  in 10 ×   2359 runs
thompson   12.79 µs ±   0.04 µs    12.74 µs …  12.84 µs  in 10 ×   7815 runs

从基准测试结果可以看出,在这种情况下虚拟机方法明显快于导数方法。导数方法需要频繁分配中间的正则表达式结构,带来了更高的开销和更慢的性能。相比之下,虚拟机执行的是一组固定的指令,一旦双端队列扩展到完整大小后,就很少需要分配新的结构了。

不过,导数方法在理论分析上更简洁。我们可以很容易地证明算法的终止性,因为需要计算的导数数量受到 AST 大小的限制,并且随着 deriv 函数的每次递归调用而严格递减。而虚拟机方法则不同,如果输入的 Prg 包含无限循环,程序可能永远不会终止,这就需要仔细处理线程优先级,以避免无限循环和线程数量的指数级增长。