跳到主要内容

prettyprinter:使用函数组合解决结构化数据打印问题

· 阅读需 9 分钟

结构化数据的打印是编程中常见的问题,尤其是在调试和日志记录时。如何展示复杂的数据结构,并能够根据屏幕宽度调整排版?例如,对于一个数组字面量 [a,b,c] , 我们希望在屏幕宽度足够时打印为一行,而在屏幕宽度不足时自动换行并缩进。 传统的解决方案往往依赖于手动处理字符串拼接和维护缩进状态,这样的方式不仅繁琐,而且容易出错。

本篇文章将介绍一种基于函数组合的实用方案——prettyprinter的实现。Prettyprinter 向用户提供了一系列函数, 这些函数能够组合成一个描述了打印方式的Doc原语。然后,根据宽度配置和Doc原语生成最终的字符串。函数组合的思路使得用户能够复用已有的代码,声明式地实现数据结构的打印。

SimpleDoc 原语

我们先定义一个SimpleDoc表示4个最简单的原语,来处理最基本的字符串拼接和换行。

enum SimpleDoc {
  Empty
  Line
  Text(String)
  Cat(SimpleDoc, SimpleDoc)
}
  • Empty: 表示空字符串
  • Line:表示换行
  • Text(String): 表示一个不包含换行的文本片段
  • Cat(SimpleDoc, SimpleDoc): 按顺序组合两个 SimpleDoc

按照上面每个原语的定义,我们可以实现一个简单的渲染函数:这个函数使用一个栈来保存待处理的SimpleDoc,逐个将它们转换为字符串。

fn SimpleDoc::render(doc : SimpleDoc) -> String {
  let buf = StringBuilder::new()
  let stack = [doc]
  while stack.pop() is Some(doc) {
    match doc {
      Empty => ()
      Line => {
        buf..write_string("\n")
      }
      Text(text) => {
        buf.write_string(text)
      }
      Cat(left, right) =>
        stack..push(right)..push(left)
    }
  }
  buf.to_string()
}

编写测试,可以看到SimpleDoc的表达能力和 String 相当: Empty 相当于 "" , Line 相当于 "\n" , Text("a") 相当于 "a" , Cat(Text("a"), Text("b")) 相当于 "a" + "b" 。

test "simple doc" {
  let doc : SimpleDoc = Cat(Text("hello"), Cat(Line, Text("world")))
  inspect(
    doc.render(),
    content=(
      #|hello
      #|world
    ),
  )
}

目前它还和String一样无法方便地处理缩进和排版切换。不过,只要再添加三个原语就可以解决这些问题。

ExtendDoc:Nest, Choice, Group

接下来我们在SimpleDoc的基础上,添加三个新的原语Nest、Choice、Group来处理更复杂的打印需求。

enum ExtendDoc {
  Empty
  Line
  Text(String)
  Cat(ExtendDoc, ExtendDoc)
  Nest(Int,ExtendDoc)
  Choice(ExtendDoc, ExtendDoc)
  Group(ExtendDoc)
}
  • Nest Nest(Int, ExtendDoc) 用于处理缩进。第一个参数表示缩进的空格数,第二个参数表示内部的 ExtendDoc 。当内部的 ExtendDoc 包含 Line 时,render函数将在打印换行的同时追加相应数量的空格。 Nest 嵌套使用时缩进会累加。

  • Choice Choice(ExtendDoc, ExtendDoc) 保存了两种打印方式。通常第一个参数表示不包含换行更紧凑的布局,第二个参数则是包含 Line 的布局。当render在紧凑模式时,使用第一个布局,否则使用第二个。

  • Group Group(ExtendDoc) 将ExtendDoc分组,并根据 ExtendDoc 的长度和剩余的空间切换打印 ExtendDoc 时的模式。如果剩余空间足够,则在紧凑模式下打印,否则使用包含换行的布局。

计算所需空间​

Group的实现需要计算 ExtendDoc 的空间需求,以便决定是否使用紧凑模式。我们可以为 ExtendDoc 添加一个 space() 方法来计算每个布局片段所需的空间。

let max_space = 9999

fn ExtendDoc::space(self : Self) -> Int {
  match self {
    Empty => 0
    Line => max_space
    Text(str) => str.length()
    Cat(a, b) => a.space() + b.space()
    Nest(_, a) | Choice(a, _) | Group(a) => a.space()
  }
}

对于 Line , 我们假设它总是需要占用无限大的空间。这样如果 Group 内包含 Line,能够保证render处理内部的 ExtendDoc 时不会进入紧凑模式。

实现 ExtendDoc::render​

我们在SimpleDoc::render的基础上实现 ExtendDoc::render 。 render在打印完一个子结构后,继续打印后续的结构需要退回到原先的缩进层级,因此需要在stack中额外保存每个待打印的ExtendDoc的两个状态:缩进和是否在紧凑模式。我们还需要维护了一个在render过程中更新的 column 变量,表示当前行的已经使用的字符数,以计算当前行所剩的空间。另外,函数增加了额外的width参数,表示每行的最大宽度限制。

fn ExtendDoc::render(doc : ExtendDoc, width~ : Int = 80) -> String {
  let buf = StringBuilder::new()
  let stack = [(0, false, doc)] // 默认不缩进,非紧凑模式
  let mut column = 0
  while stack.pop() is Some((indent, fit, doc)) {
    match doc {
      Empty => ()
      Line => {
        buf..write_string("\n")
        // 在换行后打印需要的缩进
        for _ in 0..<indent {
          buf.write_string(" ")
        }
        // 重置当前行的字符数
        column = indent
      }
      Text(text) => {
        buf.write_string(text)
        // 更新当前行的字符数
        column += text.length()
      }
      Cat(left, right) =>
        stack..push((indent, fit, right))..push((indent, fit, left))
      Nest(n, doc) => stack..push((indent + n, fit, doc)) // 增加缩进
      Choice(a, b) =>
        stack.push(if fit { (indent, fit, a) } else { (indent, fit, b) })
      Group(doc) => {
        // 如果已经在紧凑模式下,直接使用紧凑布局。如果不在紧凑模式下,但是要打印的内容可以放入当前行,则进入紧凑模式。
        let fit = fit || column + doc.space() <= width
        stack.push((indent, fit, doc))
      }
    }
  }
  buf.to_string()
}

下面我们尝试用 ExtendDoc 描述一个 (expr) ,并在不同的宽度配置下打印它:

let softline : ExtendDoc = Choice(Empty, Line)

impl Add for ExtendDoc with op_add(a, b) {
  Cat(a, b)
}

test "tuple" {
  let tuple : ExtendDoc = Group(
    Text("(") + Nest(2, softline + Text("expr")) + softline + Text(")"),
  )
  inspect(tuple.render(width=40), content="(expr)")
  inspect(
    tuple.render(width=5),
    content=(
      #|(
      #|  expr
      #|)
    ),
  )
}

我们先通过组合Empty和Line的方式定义了一个在紧凑模式下不换行的 softline 。render默认以非紧凑模式开始打印,所以我们需要用 Group 将整个表达式包裹起来。这样在宽度足够时,整个表达式会打印为一行,而在宽度不足时会自动换行并缩进。为了减少嵌套的括号,改善可读性,这里给 ExtendDoc 重载了 + 运算符。

组合函数

在prettyprinter的实践中,用户更多地会使用在 ExtendDoc 原语基础之上组合出的函数——例如之前使用过的 softline 。下面将介绍一些实用的函数,帮助我们解决结构化打印的问题。

softline & softbreak​

let softbreak : ExtendDoc = Choice(Text(" "), Line)

和 softline 类似,不同的是在紧凑模式下它会加入额外的空格。注意在同一层 Group 中,每个 Choice 都会一致选择紧凑或非紧凑模式。

let abc : ExtendDoc = Text("abc")

let def : ExtendDoc = Text("def")

let ghi : ExtendDoc = Text("ghi")

test "softbreak" {
  let doc : ExtendDoc = Group(abc + softbreak + def + softbreak + ghi)
  inspect(doc.render(width=20), content="abc def ghi")
  inspect(
    doc.render(width=10),
    content=(
      #|abc
      #|def
      #|ghi
    ),
  )
}

autoline & autobreak​

let autoline : ExtendDoc = Group(softline)

let autobreak : ExtendDoc = Group(softbreak)

autoline 和 autobreak 实现一种类似于文字编辑器的排版:尽可能多地将内容放进一行内,溢出则换行。

test {
  let doc : ExtendDoc = Group(
    abc + autobreak + def + autobreak + ghi,
  )
  inspect(doc.render(width=10), content="abc def ghi")
  inspect(
    doc.render(width=5),
    content=(
      #|abc def
      #|ghi
    ),
  )
  inspect(
    doc.render(width=3),
    content=(
      #|abc
      #|def
      #|ghi
    ),
  )
}

sepby​

fn sepby(xs : Array[ExtendDoc], sep : ExtendDoc) -> ExtendDoc {
  match xs {
    [] => Empty
    [x, .. xs] => xs.fold(init=x, (a, b) => a + sep + b)
  }
}

sepby会在ExtendDoc之间插入分隔符sep。

let comma : ExtendDoc = Text(",")
test {
  let layout = Group(sepby([abc, def, ghi], comma + softbreak))
  inspect(layout.render(width=40), content="abc, def, ghi")
  inspect(
    layout.render(width=10),
    content=(
      #|abc,
      #|def,
      #|ghi

    ),
  )
}

surround​

fn surround(m : ExtendDoc, l : ExtendDoc, r : ExtendDoc) -> ExtendDoc {
  l + m + r
}

surround 用于在 ExtendDoc 的两侧添加括号或其他分隔符。

test {
  inspect(surround(abc, Text("("), Text(")")).render(), content="(abc)")
}

打印Json

利用上面定义的函数,我们可以实现一个打印Json的函数。这个函数将递归地处理Json的每个元素,生成相应的布局。

fn pretty(x : Json) -> ExtendDoc {
  fn comma_list(xs, l, r) {
    (Nest(2, softline + sepby(xs, comma + softbreak)) + softline)
    |> surround(l, r)
    |> Group
  }

  match x {
    Array(elems) => {
      let elems = elems.iter().map(pretty).collect()
      comma_list(elems, Text("["), Text("]"))
    }
    Object(pairs) => {
      let pairs = pairs
        .iter()
        .map(p => Group(Text(p.0.escape()) + Text(": ") + pretty(p.1)))
        .collect()
      comma_list(pairs, Text("{"), Text("}"))
    }
    String(s) => Text(s.escape())
    Number(i) => Text(i.to_string())
    False => Text("false")
    True => Text("true")
    Null => Text("null")
  }
}

可以看到在不同的打印宽度下,Json的排版会自动调整。

test {
  let json : Json = {
    "key1": "string",
    "key2": [12345, 67890],
    "key3": [
      { "field1": 1, "field2": 2 },
      { "field1": 1, "field2": 2 },
      { "field1": [1, 2], "field2": 2 },
    ],
  }
  inspect(
    pretty(json).render(width=80),
    content=(
      #|{
      #|  "key1": "string",
      #|  "key2": [12345, 67890],
      #|  "key3": [
      #|    {"field1": 1, "field2": 2},
      #|    {"field1": 1, "field2": 2},
      #|    {"field1": [1, 2], "field2": 2}
      #|  ]
      #|}
    ),
  )
  inspect(
    pretty(json).render(width=30),
    content=(
      #|{
      #|  "key1": "string",
      #|  "key2": [12345, 67890],
      #|  "key3": [
      #|    {"field1": 1, "field2": 2},
      #|    {"field1": 1, "field2": 2},
      #|    {
      #|      "field1": [1, 2],
      #|      "field2": 2
      #|    }
      #|  ]
      #|}
    ),
  )
  inspect(
    pretty(json).render(width=20),
    content=(
      #|{
      #|  "key1": "string",
      #|  "key2": [
      #|    12345,
      #|    67890
      #|  ],
      #|  "key3": [
      #|    {
      #|      "field1": 1,
      #|      "field2": 2
      #|    },
      #|    {
      #|      "field1": 1,
      #|      "field2": 2
      #|    },
      #|    {
      #|      "field1": [
      #|        1,
      #|        2
      #|      ],
      #|      "field2": 2
      #|    }
      #|  ]
      #|}
    ),
  )
}

总结

本文介绍了如何简单实现一个prettyprinter,使用函数组合的方式来处理结构化数据的打印。通过定义一系列原语和组合函数,我们可以灵活地控制打印格式,并根据屏幕宽度自动调整布局。

当前的实现还可以进一步优化,例如通过记忆化space的计算来提高性能。ExtendDoc::render函数可以增加一个ribbon参数,分别统计当前行的空格和其他文本字数,并且在Group的紧凑模式判断中增加额外的条件,来控制每行的信息密度。另外,还可以增加更多的原语来实现悬挂缩进、最小换行数量等功能。对于更多的设计和实现细节感兴趣的读者,可以参考A prettier printer - Philip Wadler以及Haskell、OCaml等语言的prettyprinter实现。

Mini-adapton: 用 MoonBit 实现增量计算

· 阅读需 10 分钟

介绍​

让我们先用一个类似 excel 的例子感受一下增量计算长什么样子. 首先, 定义一个这样的依赖图:

在这个图中, t1 的值通过 n1 + n2 计算得到, t2 的值通过 t1 + n3 计算得到.

当我们想得到 t2 的值时, 该图定义的计算将被执行: 首先通过 n1 + n2 算出 t1, 再通过 t1 + n3 算出 t2. 这个过程和非增量计算是相同的.

但当我们开始改变n1, n2 或 n3 的值时, 事情就不一样了. 比如说我们想将 n1 和 n2 的值互换, 再得到 t2 的值. 在非增量计算中, t1 和 t2 都将被重新计算一遍, 但实际上 t2 是不需要被重新计算的, 因为它依赖的两个值 t1 和 n3 都没有改变 (将 n1 和 n2 的值互换不会改变 t1 的值).

下面的代码实现了我们刚刚举的例子. 我们使用 Cell::new 来定义 n1, n2 和 n3 这些不需要计算的东西, 使用 Thunk::new 来定义 t1 和 t2 这样需要计算的东西.

test {
  // a counter to record the times of t2's computation
  let mut cnt = 0
  // start define the graph
  let n1 = Cell::new(1)
  let n2 = Cell::new(2)
  let n3 = Cell::new(3)
  let t1 = Thunk::new(fn() {
    n1.get() + n2.get()
  })
  let t2 = Thunk::new(fn() {
    cnt += 1
    t1.get() + n3.get()
  })
  // get the value of t2
  inspect(t2.get(), content="6")
  inspect(cnt, content="1")
  // swap value of n1 and n2
  n1.set(2)
  n2.set(1)
  inspect(t2.get(), content="6")
  // t2 does not recompute
  inspect(cnt, content="1")
}

在这篇文章中, 我们将介绍如何在 MoonBit 中实现一个增量计算库. 这个库的 API 就是我们上面例子中出现的那些:

Cell::new
Cell::get
Cell::set
Thunk::new
Thunk::get

问题分析和解法​

要实现这个库, 我们主要有三个问题需要解决:

如何在运行时构建依赖图​

作为一个使用 MoonBit 实现的库, 没有简单方法让我们可以静态地构建依赖图, 因为 MoonBit 目前还不支持任何元编程的机制. 因此我们需要动态地把依赖图构建出来. 事实上, 我们关心的只是哪些 thunk 或 cell 被另一个 thunk 依赖了, 所以一个不错的构建依赖图的时机就是在用户调用 Thunk::get 的时候. 比如在上面的例子中:

let n1 = Cell::new(1)
let n2 = Cell::new(2)
let n3 = Cell::new(3)
let t1 = Thunk::new(fn() { n1.get() + n2.get() })
let t2 = Thunk::new(fn() { t1.get() + n3.get() })
t2.get()

当用户调用 t2.get() 时, 我们在运行时会知道 t1.get() 和 n3.get() 在其中也被调用了. 因此 t1 和 n3 是 t2 的依赖, 并且我们可以构建一个这样的图:

同样的过程也会在 t1.get() 被调用时发生.

所以计划是这样的:

  1. 我们定义一个栈来记录我们当前在获得哪个 thunk 的值. 在这里使用栈的原因是, 我们事实上是在尝试记录每个 get 的调用栈.
  2. 当我们调用 get 时, 将其标记为栈顶 thunk 的依赖, 如果它是一个 thunk, 再把它压栈.
  3. 当一个 thunk 的 get 结束时, 将它出栈.

让我们看看上面那个例子在这个算法下的过程是什么样子的:

  1. 当我们调用 t2.get 时, 将 t2 压栈.

  2. 当我们在 t2.get 中调用 t1.get 时, 将 t1 记为 t2 的依赖, 并将 t1 压栈.

  3. 当我们在 t1.get 中调用 n1.get 时, 将 n1 记为 t1 的依赖

  4. 相同的过程发生在 n2 身上.

  5. 当 t1.get 结束时, 将 t1 出栈.

  6. 当我们调用 n3.get 时, 将 n3 记为 t2 的依赖.

除了这些从父依赖到子依赖的边之外, 我们最好也记录一个从子依赖到父依赖的边, 方便后面我们在这个图上反向便利.

在接下来的代码中, 我们将使用 outgoing_edges 指代从父依赖到子依赖的边, 使用 incoming_edges 指代中子依赖到父依赖的边.

如何标记过时的节点​

当我们调用 Cell::set 时, 该节点本身和所有依赖它的节点都应该被标记为过时的. 这将在后面作为判断一个 thunk 是否需要重新计算的标准之一. 这基本上是一个从图的叶子节点向后遍历的过程. 我们可以用这样的伪 MoonBit 代码表示这个算法:

fn dirty(node: Node) -> Unit {
  for n in node.incoming_edges {
    n.set_dirty(true)
    dirty(node)
  }
}

如何决定一个 thunk 需要被重新计算​

当我们调用 Thunk::get 时, 我们需要决定是否它需要被重新计算. 但只用我们在上一节描述的方法是不够的. 如果我们只使用是否过时这一个标准进行判断, 势必会有不需要的计算发生. 比如我们在一开始给出的例子:

n1.set(2)
n2.set(1)
inspect(t2.get(), content="6")

当我们调换 n1 和 n2 的值时, n1, n2, t1 和 t2 都应该被标记为过时, 但当我们调用 t2.get 时, 其实没有必要重新计算 t2, 因为 t1 的值并没有改变.

这提醒我们除了过时之外, 我们还要考虑依赖的值是否和它上一次的值一样. 如果一个节点既是过时的, 并且它的依赖中存在一个值和上一次不同, 那么它应该被重新计算.

我们可以用下面的伪 MoonBit 代码描述这个算法:

fn propagate(self: Node) -> Unit {
  // 当一个节点过时了, 它可能需要被重新计算
  if self.is_dirty() {
    // 重新计算之后, 它将不在是过时的
    self.set_dirty(false)
    for dependency in self.outgoing_edges() {
      // 递归地重新计算每个依赖
      dependency.propagate()
      // 如果一个依赖的值改变了, 这个节点需要被重新计算
      if dependency.is_changed() {
        // 移除所有的 outgoing_edges, 它们将在被计算时重新构建
        self.outgoing_edges().clear()
        self.evaluate()
        return
      }
    }
  }
}

实现​

基于上面描述的代码, 实现是比较直观的.

首先, 我们先定义 Cell:

struct Cell[A] {
  mut is_dirty : Bool
  mut value : A
  mut is_changed : Bool
  incoming_edges : Array[&Node]
}

由于 Cell 只会是依赖图中的叶子节点, 所以它没有 outgoing_edges. 这里出现的特征 Node 是用来抽象依赖图中的节点的.

接着, 我们定义 Thunk:

struct Thunk[A] {
  mut is_dirty : Bool
  mut value : A?
  mut is_changed : Bool
  thunk : () -> A
  incoming_edges : Array[&Node]
  outgoing_edges : Array[&Node]
}

Thunk 的值是可选的, 因为它只有在我们第一次调用 Thunk::get 之后才会存在.

我们可以很简单地给这两个类型实现 new:

fn[A : Eq] Cell::new(value : A) -> Cell[A] {
  Cell::{
    is_changed: false,
    value,
    incoming_edges: [],
    is_dirty: false,
  }
}
fn[A : Eq] Thunk::new(thunk : () -> A) -> Thunk[A] {
  Thunk::{
    value: None,
    is_changed: false,
    thunk,
    incoming_edges: [],
    outgoing_edges: [],
    is_dirty: false,
  }
}

Thunk 和 Cell 是依赖图的两种节点, 我们可以使用一个特征 Node 来抽象它们:

trait Node {
  is_dirty(Self) -> Bool
  set_dirty(Self, Bool) -> Unit
  incoming_edges(Self) -> Array[&Node]
  outgoing_edges(Self) -> Array[&Node]
  is_changed(Self) -> Bool
  evaluate(Self) -> Unit
}

为两个类型实现这个特征:

impl[A] Node for Cell[A] with incoming_edges(self) {
  self.incoming_edges
}

impl[A] Node for Cell[A] with outgoing_edges(_self) {
  []
}

impl[A] Node for Cell[A] with is_dirty(self) {
  self.is_dirty
}

impl[A] Node for Cell[A] with set_dirty(self, new_dirty) {
  self.is_dirty = new_dirty
}

impl[A] Node for Cell[A] with is_changed(self) {
  self.is_changed
}

impl[A] Node for Cell[A] with evaluate(_self) {
  ()
}

impl[A : Eq] Node for Thunk[A] with is_changed(self) {
  self.is_changed
}

impl[A : Eq] Node for Thunk[A] with outgoing_edges(self) {
  self.outgoing_edges
}

impl[A : Eq] Node for Thunk[A] with incoming_edges(self) {
  self.incoming_edges
}

impl[A : Eq] Node for Thunk[A] with is_dirty(self) {
  self.is_dirty
}

impl[A : Eq] Node for Thunk[A] with set_dirty(self, new_dirty) {
  self.is_dirty = new_dirty
}

impl[A : Eq] Node for Thunk[A] with evaluate(self) {
  node_stack.push(self)
  let value = (self.thunk)()
  self.is_changed = match self.value {
    None => true
    Some(v) => v != value
  }
  self.value = Some(value)
  node_stack.unsafe_pop() |> ignore
}

这里唯一复杂的实现是 Thunk 的 evaluate. 这里我们需要先把这个 thunk 推到栈顶用于后面的依赖记录. node_stack 的定义如下:

let node_stack : Array[&Node] = []

然后做真正的计算, 并且把计算得到的值和上一个值做比较以更新 self.is_changed. is_changed 会在后面帮助我们判断是否需要重新计算一个 thunk.

dirty 和 propagate 的实现几乎和上面的伪代码相同:

fn &Node::dirty(self : &Node) -> Unit {
  for dependent in self.incoming_edges() {
    if not(dependent.is_dirty()) {
      dependent.set_dirty(true)
      dependent.dirty()
    }
  }
}
fn &Node::propagate(self : &Node) -> Unit {
  if self.is_dirty() {
    self.set_dirty(false)
    for dependency in self.outgoing_edges() {
      dependency.propagate()
      if dependency.is_changed() {
        self.outgoing_edges().clear()
        self.evaluate()
        return
      }
    }
  }
}

有了这些函数的帮助, 最主要的三个 API: Cell::get, Cell::set 和 Thunk::get 实现起来就比较简单了.

为了得到一个 cell 的值, 我们直接返回结构体的 value 字段即可. 但在此之前, 如果它是在一个 Thunk::get 中被调用的, 我们要先把他记录为依赖.

fn[A] Cell::get(self : Cell[A]) -> A {
  if node_stack.last() is Some(target) {
    target.outgoing_edges().push(self)
    self.incoming_edges.push(target)
  }
  self.value
}

当我们更改一个 cell 的值时, 我们需要先确保 is_changed 和 dirty 这两个状态被正确地更新了, 再将它的每一个父依赖标记为过时.

fn[A : Eq] Cell::set(self : Cell[A], new_value : A) -> Unit {
  if self.value != new_value {
    self.is_changed = true
    self.value = new_value
    self.set_dirty(true)
    &Node::dirty(self)
  }
}

和 Cell::get 类似, 在实现 Thunk::get 时我们需要先将 self 记录为依赖. 之后我们模式匹配 self.value, 如果它是 None, 这意味着这是第一次用户尝试计算这个 thunk 地值, 我们可以简单地直接计算它; 如果它是 Some, 我们需要使用 propagate 来确保我们只重新计算那些需要的 thunk.

fn[A : Eq] Thunk::get(self : Thunk[A]) -> A {
  if node_stack.last() is Some(target) {
    target.outgoing_edges().push(self)
    self.incoming_edges.push(target)
  }
  match self.value {
    None => self.evaluate()
    Some(_) => &Node::propagate(self)
  }
  self.value.unwrap()
}

参考​

MoonBit与Python集成指南

· 阅读需 13 分钟

引言​

Python,以其简洁的语法和庞大的生态系统,已成为当今最受欢迎的编程语言之一。然而,围绕其性能瓶颈和动态类型系统在大型项目中的维护性问题的讨论也从未停止。为了解决这些挑战,开发者社区探索了多种优化路径。

MoonBit 官方推出的 python.mbt 工具为此提供了一个新的视角。它允许开发者在 MoonBit 环境中直接调用 Python 代码。这种结合旨在融合 MoonBit 的静态类型安全、高性能潜力与 Python 成熟的生态系统。通过 python.mbt,开发者可以在享受 Python 丰富库函数的同时,利用 MoonBit 的静态分析能力、现代化的构建与测试工具,为构建大规模、高性能的系统级软件提供可能。

本文旨在深入探讨 python.mbt 的工作原理,并提供一份实践指南。本文将解答一些常见问题,例如:python.mbt 如何工作?它是否会因为增加了一个中间层而比原生 Python 更慢?相较于 C++ 的 pybind11 或 Rust 的 PyO3 等现有工具,python.mbt 的优势何在?要回答这些问题,我们首先需要理解 Python 解释器的基本工作流程。

Python 解释器的工作原理​

Python 解释器执行代码主要经历三个阶段:

  1. 解析阶段 (Parsing) :此阶段包含词法分析和语法分析。解释器将人类可读的 Python 源代码分解成一个个标记(Token),然后根据语法规则将这些标记组织成一个树形结构,即抽象语法树(AST)。

    例如,对于以下 Python 代码:

    def add(x, y):
      return x + y
    
    a = add(1, 2)
    print(a)

    我们可以使用 Python 的 ast 模块来查看其生成的 AST 结构:

    Module(
        body=[
            FunctionDef(
                name='add',
                args=arguments(
                    args=[
                        arg(arg='x'),
                        arg(arg='y')]),
                body=[
                    Return(
                        value=BinOp(
                            left=Name(id='x', ctx=Load()),
                            op=Add(),
                            right=Name(id='y', ctx=Load())))]),
            Assign(
                targets=[
                    Name(id='a', ctx=Store())],
                value=Call(
                    func=Name(id='add', ctx=Load()),
                    args=[
                        Constant(value=1),
                        Constant(value=2)])),
            Expr(
                value=Call(
                    func=Name(id='print', ctx=Load()),
                    args=[
                        Name(id='a', ctx=Load())]))])
  2. 编译阶段 (Compilation) :接下来,Python 解释器会将 AST 编译成更低级、更线性的中间表示,即字节码(Bytecode)。这是一种平台无关的指令集,专为 Python 虚拟机(PVM)设计。

    利用 Python 的 dis 模块,我们可以查看上述代码对应的字节码:

      2           LOAD_CONST               0 (<code object add>)
                  MAKE_FUNCTION
                  STORE_NAME               0 (add)
    
      5           LOAD_NAME                0 (add)
                  PUSH_NULL
                  LOAD_CONST               1 (1)
                  LOAD_CONST               2 (2)
                  CALL                     2
                  STORE_NAME               1 (a)
    
      6           LOAD_NAME                2 (print)
                  PUSH_NULL
                  LOAD_NAME                1 (a)
                  CALL                     1
                  POP_TOP
                  RETURN_CONST             3 (None)
  3. 执行阶段 (Execution) :最后,Python 虚拟机(PVM)会逐条执行字节码指令。每条指令都对应 CPython 解释器底层的一个 C 函数调用。例如,LOAD_NAME 会查找变量,BINARY_OP 会执行二元运算。正是这个逐条解释执行的过程,构成了 Python 性能开销的主要来源。一次简单的 1 + 2 运算,背后需要经历整个解析、编译和虚拟机执行的复杂流程。

了解这个流程,有助于我们理解 Python 性能优化的基本思路,以及 python.mbt 的设计哲学。

优化 Python 性能的路径​

目前,提升 Python 程序性能主要有两种主流方法:

  1. 即时编译(JIT) 。像 PyPy 这样的项目,通过分析正在运行的程序,将频繁执行的"热点"字节码编译成高度优化的本地机器码,从而绕过 PVM 的解释执行,大幅提升计算密集型任务的速度。然而,JIT 并非万能药,它无法解决 Python 动态类型语言的固有问题,例如在大型项目中难以进行有效的静态分析,这给软件维护带来了挑战。
  2. 原生扩展。开发者可以使用 C++(借助 pybind11)或 Rust(借助 PyO3)等语言直接调用Python功能,或者用这些语言来编写性能关键模块,然后从 Python 中调用。这种方法可以获得接近原生的性能,但它要求开发者同时精通 Python 和一门复杂的系统级语言,学习曲线陡峭,对大多数 Python 程序员来说门槛较高。

python.mbt 也是一种原生扩展。但相比较于C++和Rust等语言,它试图在性能、易用性和工程化能力之间找到一个新的平衡点,更强调在MoonBit语言中直接使用Python功能。

  1. 高性能核心:MoonBit 是一门静态类型的编译型语言,其代码可以被高效地编译成原生机器码。开发者可以将计算密集型逻辑用 MoonBit 实现,从根本上获得高性能。
  2. 无缝的 Python 调用:python.mbt 直接与 CPython 的 C-API 交互,调用 Python 模块和函数。这意味着调用开销被最小化,绕过了 Python 的解析和编译阶段,直达虚拟机执行层。
  3. 更平缓的学习曲线:相较于 C++ 和 Rust,MoonBit 的语法设计更加现代化和简洁,并拥有完善的函数式编程支持、文档系统、单元测试和静态分析工具,对习惯于 Python 的开发者更加友好。
  4. 改善的工程化与 AI 协作:MoonBit 的强类型系统和清晰的接口定义,使得代码意图更加明确,更易于被静态分析工具和 AI 辅助编程工具理解。这有助于在大型项目中维护代码质量,并提升与 AI 协作编码的效率和准确性。

在 MoonBit 中使用已封装的 Python 库​

为了方便开发者使用,MoonBit 官方会在构建系统和IDE成熟后对主流 Python 库进行封装。封装完成后,用户可以像导入普通 MoonBit 包一样,在项目中使用这些 Python 库。下面以 matplotlib 绘图库为例。

首先,在你的项目根目录的 moon.pkg.json 或终端中添加 matplotlib 依赖:

moon update
moon add Kaida-Amethyst/matplotlib

然后,在要使用该库的子包的 moon.pkg.json 中声明导入。这里,我们遵循 Python 的惯例,为其设置一个别名 plt:

{
  "import": [
    {
      "path": "Kaida-Amethyst/matplotlib",
      "alias": "plt"
    }
  ]
}

完成配置后,便可以在 MoonBit 代码中调用 matplotlib 进行绘图:

let sin : (Double) -> Double = @math.sin

fn main {
  let x = Array::makei(100, fn(i) { i.to_double() * 0.1 })
  let y = x.map(sin)

  // 为保证类型安全,封装后的 subplots 接口总是返回一个固定类型的元组。
  // 这避免了 Python 中根据参数返回不同类型对象的动态行为。
  let (_, axes) = plt::subplots(1, 1)

  // 使用 .. 级联调用语法
  axes[0][0]
  ..plot(x, y, color = Green, linestyle = Dashed, linewidth = 2)
  ..set_title("Sine of x")
  ..set_xlabel("x")
  ..set_ylabel("sin(x)")

  @plt.show()
}

目前,在 macOS 和 Linux 环境下,MoonBit 的构建系统可以自动处理依赖。在 Windows 上,用户可能需要手动安装 C 编译器并配置 Python 环境。未来的 MoonBit IDE 将致力于简化这一过程。

在 MoonBit 中使用未封装的 Python 模块​

Python 生态浩如烟海,即使现在有了AI技术,完全依赖官方封装也并不现实。幸运的是,我们可以利用 python.mbt 的核心功能直接与任何 Python 模块交互。下面,我们以 Python 标准库中,一个简单的的 time 模块为例,演示这一过程。

引入 python.mbt​

首先,确保你的 MoonBit 工具链是最新版本,然后添加 python.mbt 依赖:

moon update
moon add Kaida-Amethyst/python

接着,在你的包的 moon.pkg.json 中导入它:

{
  "import": ["Kaida-Amethyst/python"]
}

python.mbt 会自动处理 Python 解释器的初始化(Py_Initialize)和关闭,开发者无需手动管理。

导入 Python 模块​

使用 @python.pyimport 函数来导入模块。为了避免重复导入造成的性能损耗,建议使用闭包技巧来缓存导入的模块对象:

// 定义一个结构体来持有 Python 模块对象,增强类型安全
pub struct TimeModule {
  time_mod: PyModule
}

// 定义一个函数,它返回一个闭包,该闭包用于获取 TimeModule 实例
fn import_time_mod() -> () -> TimeModule {
  // 仅在首次调用时执行导入操作
  guard @python.pyimport("time") is Some(time_mod) else {
    println("Failed to load Python module: time")
    panic("ModuleLoadError")
  }
  let time_mod = TimeModule::{ time_mod }
  // 返回的闭包会捕获 time_mod 变量
  fn () { time_mod }
}

// 创建一个全局的 time_mod "getter" 函数
let time_mod: () -> TimeModule = import_time_mod()

在后续代码中,我们应始终通过调用 time_mod() 来获取模块,而不是 import_time_mod。

MoonBit 与 Python 对象的相互转换​

要调用 Python 函数,我们需要在 MoonBit 对象和 Python 对象(PyObject)之间进行转换。

  1. 整数: 使用 PyInteger::from 从 Int64 创建 PyInteger,使用 to_int64() 反向转换。

    test "py_integer_conversion" {
      let n: Int64 = 42
      let py_int = PyInteger::from(n)
      inspect(py_int, content="42")
      assert_eq(py_int.to_int64(), 42L)
    }
  2. 浮点数: 使用 PyFloat::from 和 to_double。

    test "py_float_conversion" {
      let n: Double = 3.5
      let py_float = PyFloat::from(n)
      inspect(py_float, content="3.5")
      assert_eq(py_float.to_double(), 3.5)
    }
  3. 字符串: 使用 PyString::from 和 to_string。

    test "py_string_conversion" {
      let py_str = PyString::from("hello")
      inspect(py_str, content="'hello'")
      assert_eq(py_str.to_string(), "hello")
    }
  4. 列表 (List) : 你可以创建一个空 PyList 然后 append 元素,或者直接从一个 Array[&IsPyObject] 创建。

    test "py_list_from_array" {
      let one = PyInteger::from(1)
      let two = PyFloat::from(2.0)
      let three = PyString::from("three")
      let arr: Array[&IsPyObject] = [one, two, three]
    
      let list = PyList::from(arr)
      inspect(list, content="[1, 2.0, 'three']")
    }
  5. 元组 (Tuple) : PyTuple 需要先指定大小,然后通过 set 方法逐一填充元素。

    test "py_tuple_creation" {
      let tuple = PyTuple::new(3)
      tuple
      ..set(0, PyInteger::from(1))
      ..set(1, PyFloat::from(2.0))
      ..set(2, PyString::from("three"))
    
      inspect(tuple, content="(1, 2.0, 'three')")
    }
  6. 字典 (Dict) : PyDict 主要支持字符串作为键。使用 new 创建字典,set 添加键值对。对于非字符串键,需要使用 set_by_obj。

    test "py_dict_creation" {
      let dict = PyDict::new()
      dict
      ..set("one", PyInteger::from(1))
      ..set("two", PyFloat::from(2.0))
    
      inspect(dict, content="{'one': 1, 'two': 2.0}")
    }

从 Python 复合类型中获取元素时,python.mbt 会进行运行时类型检查,并返回一个 Optional[PyObjectEnum],以确保类型安全。

test "py_list_get" {
  let list = PyList::new()
  list.append(PyInteger::from(1))
  list.append(PyString::from("hello"))

  inspect(list.get(0).unwrap(), content="PyInteger(1)")
  inspect(list.get(1).unwrap(), content="PyString('hello')")
  inspect(list.get(2), content="None") // 索引越界返回 None
}

调用模块中的函数​

调用函数分为两步:首先用 get_attr 获取函数对象,然后用 invoke 执行调用。invoke 的返回值是一个需要进行模式匹配和类型转换的 PyObject。

下面是 time.sleep 和 time.time 的 MoonBit 封装:

// 封装 time.sleep
pub fn sleep(seconds: Double) -> Unit {
  let lib = time_mod()
  guard lib.time_mod.get_attr("sleep") is Some(PyCallable(f)) else {
    println("get function `sleep` failed!")
    panic()
  }
  let args = PyTuple::new(1)
  args.set(0, PyFloat::from(seconds))
  match (try? f.invoke(args)) {
    Ok(_) => Ok(())
    Err(e) => {
      println("invoke `sleep` failed!")
      panic()
    }
  }
}

// 封装 time.time
pub fn time() -> Double {
  let lib = time_mod()
  guard lib.time_mod.get_attr("time") is Some(PyCallable(f)) else {
    println("get function `time` failed!")
    panic()
  }
  match (try? f.invoke()) {
    Ok(Some(PyFloat(t))) => t.to_double()
    _ => {
      println("invoke `time` failed!")
      panic()
    }
  }
}

完成封装后,我们就可以在 MoonBit 中以类型安全的方式使用它们了:

test "sleep" {
  let start = time().unwrap()
  sleep(1)
  let end = time().unwrap()

  println("start = \{start}")
  println("end = \{end}")
}

实践建议​

  1. 明确边界:将 python.mbt 视为连接 MoonBit 和 Python 生态的"胶水层"。将核心计算和业务逻辑保留在 MoonBit 中以利用其性能和类型系统优势,仅在必要情况下,需要调用 Python 独有库时才使用 python.mbt。

  2. 用 ADT 替代字符串魔法:许多 Python 函数接受特定的字符串作为参数来控制行为。在 MoonBit 封装中,应将这些"魔法字符串"转换为代数数据类型(ADT) ,即枚举。这利用了 MoonBit 的类型系统,将运行时的值检查提前到编译时,极大地增强了代码的健壮性。

  3. 完善的错误处理:本文中的示例为了简洁使用了 panic 或返回简单字符串。在生产代码中,应定义专门的错误类型,并通过 Result 类型进行传递和处理,提供清晰的错误上下文。

  4. 映射关键字参数:Python 函数广泛使用关键字参数(kwargs),如 plot(color='blue', linewidth=2)。这可以优雅地映射到 MoonBit 的标签参数(Labeled Arguments) 。在封装时,应优先使用标签参数以提供相似的开发体验。

    例如,一个接受 kwargs 的 Python 函数:

    # graphics.py
    def draw_line(points, color="black", width=1):
        # ... drawing logic ...
        print(f"Drawing line with color {color} and width {width}")

    其 MoonBit 封装可以设计成:

    fn draw_line(points: Array[Point], color~: Color = Black, width: Int = 1) -> Unit {
      let points : PyList = ... // convert Array[Point] to PyList
    
      // 构造args
      let args = PyTuple::new(1)
      args .. set(0, points)
    
      // 构造kwargs
      let kwargs = PyDict::new()
      kwargs
      ..set("color", PyString::from(color))
      ...set("width", PyInteger::from(width))
      match (try? f.invoke(args~, kwargs~)) {
        Ok(_) => ()
        _ => {
          // 进行错误处理
        }
      }
    }
  5. 警惕动态性:始终牢记 Python 是动态类型的。从 Python 获取的任何数据都应被视为"不可信"的,必须进行严格的类型检查和校验,尽量避免使用 unwrap,而是通过模式匹配来安全地处理所有可能的情况。

结语​

本文梳理了 python.mbt 的工作原理,并展示了如何利用它在 MoonBit 中调用 Python 代码,无论是通过预封装的库还是直接与 Python 模块交互。python.mbt 不仅仅是一个工具,它代表了一种融合思想:将 MoonBit 的静态分析、高性能和工程化优势与 Python 庞大而成熟的生态系统相结合。我们希望这篇文章能为 MoonBit 和 Python 社区的开发者们在构建未来软件时,提供一个新的、更强大的选择。

MoonBit C-FFI 开发指南

· 阅读需 17 分钟


引言​

MoonBit 是一门现代化函数式编程语言,它有着严谨的类型系统,高可读性的语法,以及专为AI设计的工具链等。然而,重复造轮子并不可取。无数经过时间检验、性能卓越的库是用C语言(或兼容C ABI的语言,如C++、Rust)编写的。从底层硬件操作到复杂的科学计算,再到图形渲染,C的生态系统是一座蕴藏着无尽宝藏的富矿。

那么,我们能否让现代的MoonBit与这些经典的C库协同工作,让新世界的开拓者也能使用旧时代的强大工具呢?答案是肯定的。通过C语言外部函数接口(C Foreign Function Interface, C-FFI),MoonBit拥有调用C函数的能力,将新旧两个世界连接起来。

这篇文章将作为你的向导,带你一步步探索MoonBit C-FFI的奥秘。我们将通过一个具体的例子——为一个C语言编写的数学库 mymath​ 创建MoonBit绑定——来学习如何处理不同类型的数据、指针、结构体乃至函数指针。

预先准备​

要连接到任何一个C库,我们需要知道这个C库的头文件的函数,如何找到头文件,如何找到库文件。对于我们这篇文章的任务来说。C语言数学库的头文件就是 mymath.h​。它定义了我们希望在MoonBit中调用的各种函数和类型。我们这里假设我们的mymath​是安装到系统上的,编译时使用-I/usr/inluclude​来找到头文件,使用-L/usr/lib -lmymath​来链接库,下面是我们的mymath.h​的部分内容。

// mymath.h

// --- 基础函数 ---
void print_version();
int version_major();
int is_normal(double input);

// --- 浮点数计算 ---
float sinf(float input);
float cosf(float input);
float tanf(float input);
double sin(double input);
double cos(double input);
double tan(double input);

// --- 字符串与指针 ---
int parse_int(char* str);
char* version();
int tan_with_errcode(double input, double* output);

// --- 数组操作 ---
int sin_array(int input_len, double* inputs, double* outputs);
int cos_array(int input_len, double* inputs, double* outputs);
int tan_array(int input_len, double* inputs, double* outputs);

// --- 结构体与复杂类型 ---
typedef struct {
  double real;
  double img;
} Complex;

Complex* new_complex(double r, double i);
void multiply(Complex* a, Complex* b, Complex** result);
void init_n_complexes(int n, Complex** complex_array);

// --- 函数指针 ---
void for_each_complex(int n, Complex** arr, void (*call_back)(Complex*));

基础准备 (The Groundwork)​

在编写任何 FFI 代码之前,我们需要先搭建好 MoonBit 与 C 代码之间的桥梁。

编译到 Native​

首先,MoonBit 代码需要被编译成原生机器码。这可以通过以下命令完成:

moon build --target native

这个命令会将你的 MoonBit 项目编译成 C 代码,并使用系统上的 C 编译器(如 GCC 或 Clang)将其编译为最终的可执行文件。编译后的 C 文件位于 target/native/release/build/​ 目录下,按包名存放在相应的子目录中。例如,main/main.mbt​ 会被编译到 target/native/release/build/main/main.c​。

配置链接​

仅仅编译是不够的,我们还需要告诉 MoonBit 编译器如何找到并链接到我们的 mymath​ 库。这需要在项目的 moon.pkg.json​ 文件中进行配置。

{
  "supported-targets": ["native"],
  "link": {
    "native": {
      "cc": "clang",
      "cc-flags": "-I/usr/include",
      "cc-link-flags": "-L/usr/lib -lmymath"
    }
  }
}
  • ​cc​: 指定用于编译C代码的编译器,例如 clang​ 或 gcc​。
  • ​cc-flags​: 编译C文件时需要的标志,通常用来指定头文件搜索路径(-I​)。
  • ​cc-link-flags​: 链接时需要的标志,通常用来指定库文件搜索路径(-L​)和具体要链接的库(-l​)。

同时,我们还需要一个 "胶水" C 文件,我们这里命名为 cwrap.c​,用来包含 C 库的头文件和 MoonBit 的运行时头文件。

// cwrap.c
#include <mymath.h>
#include <moonbit.h>

这个胶水文件也需要通过 moon.pkg.json​ 告知 MoonBit 编译器:

{
  // ... 其他配置
  "native-stub": ["cwrap.c"]
}

完成这些配置后,我们的项目就已经准备好与 mymath​ 库进行链接了。

第一次跨语言调用 (The First FFI Call)​

万事俱备,让我们来进行第一次真正的跨语言调用。在 MoonBit 中声明一个外部 C 函数,语法如下:

extern "C" fn moonbit_function_name(arg: Type) -> ReturnType = "c_function_name"
  • ​extern "C"​:告诉 MoonBit 编译器,这是一个外部 C 函数。
  • ​moonbit_function_name​:在 MoonBit 代码中使用的函数名。
  • ​"c_function_name"​:实际链接到的 C 函数的名称。

让我们用 mymath.h​ 中最简单的 version_major​ 函数来小试牛刀:

extern "C" fn version_major() -> Int = "version_major"

注意:MoonBit 拥有强大的死代码消除(DCE)能力。如果你只是声明了上面的 FFI 函数但从未在代码中(例如 main​ 函数)实际调用它,编译器会认为它是无用代码,并不会在最终生成的 C 代码中包含它的声明。所以,请确保你至少在一个地方调用了它!

跨越类型系统的鸿沟 (Navigating the Type System Chasm)​

真正的挑战在于处理两种语言之间的数据类型差异,对于一些复杂的类型情况,需要读者有一定的C语言知识。

3.1 基本类型:(Basic Types)​

对于基础的数值类型,MoonBit 和 C 之间有直接且清晰的对应关系。

MoonBit TypeC TypeNotes
​Int​​int32_t​
​Int64​​int64_t​
​UInt​​uint32_t​
​UInt64​​uint64_t​
​Float​​float​
​Double​​double​
​Bool​​int32_t​C语言标准没有原生 bool​,通常用 int32_t​ (0/1) 表示
​Unit​​void​ (返回值)用于表示 C 函数没有返回值的情况
​Byte​​uint8_t​

根据这个表格,我们可以轻松地为 mymath.h​ 中的大部分简单函数编写 FFI 声明:

extern "C" fn print_version() -> Unit = "print_version"
extern "C" fn version_major() -> Int = "version_major"

// 返回值语义上是布尔值,使用 MoonBit 的 Bool 类型更清晰
extern "C" fn is_normal(input: Double) -> Bool = "is_normal"

extern "C" fn sinf(input: Float) -> Float = "sinf"
extern "C" fn cosf(input: Float) -> Float = "cosf"
extern "C" fn tanf(input: Float) -> Float = "tanf"

extern "C" fn sin(input: Double) -> Double = "sin"
extern "C" fn cos(input: Double) -> Double = "cos"
extern "C" fn tan(input: Double) -> Double = "tan"

3.2 字符串 (Strings)​

事情在遇到字符串时开始变得有趣。你可能会想当然地把 C 的 char*​ 映射到 MoonBit 的 String​,但这是一个常见的陷阱。

​MoonBit​ 的 String​ 和 C 的 char*​ 在内存布局上完全不同。char*​ 是一个指向以 \0​ 结尾的字节序列的指针,而 MoonBit​ 的 String​ 是一个由 GC 管理的、包含长度信息和 UTF-16 编码数据的复杂对象。

参数传递:从 MoonBit 到 C

当我们需要将一个 MoonBit 字符串传递给一个接受 char*​ 的 C 函数时(如 parse_int​),我们需要手动进行转换。一个推荐的做法是将其转换为 Bytes​ 类型。

// 一个辅助函数,将 MoonBit String 转换为 C 期望的 null-terminated byte array
fn string_to_c_bytes(s: String) -> Bytes {
  let mut arr = s.to_bytes().to_array()
  // 确保以 \0 结尾
  if arr.last() != Some(0) {
    arr.push(0)
  }
  Bytes::from_array(arr)
}

// FFI 声明,注意参数类型是 Bytes
#borrow(s) // 告诉编译器我们只是借用 s,不要增加其引用计数
extern "C" fn __parse_int(s: Bytes) -> Int = "parse_int"

// 封装成一个对用户友好的 MoonBit 函数
fn parse_int(str: String) -> Int {
  let s = string_to_c_bytes(str)
  __parse_int(s)
}

​ ​#borrow​​ 标记 ​borrow​ 标记是一个优化提示。它告诉编译器,C函数只是"借用"这个参数,不会持有它的所有权。这可以避免不必要的引用计数操作,防止潜在的内存泄漏。

返回值:从 C 到 MoonBit

反过来,当 C 函数返回一个 char*​ 时(如 version​),情况更加复杂。我们绝对不能直接将其声明为返回 Bytes​ 或 String​:

// 错误的做法!
extern "C" fn version() -> Bytes = "version"

这是因为 C 函数返回的只是一个裸指针,它缺少 MoonBit GC 所需的头部信息。直接这样转换会导致运行时崩溃。

正确的做法是,将返回的 char*​ 视为一个不透明的句柄,然后在 C "胶水" 代码中编写一个转换函数,手动将其转换为一个合法的 MoonBit 字符串。

MoonBit 侧:

// 1. 声明一个外部类型来代表 C 字符串指针
#extern
type CStr

// 2. 声明一个 FFI 函数,它调用 C 包装器
extern "C" fn CStr::to_string(self: Self) -> String = "cstr_to_moonbit_str"

// 3. 声明原始的 C 函数,它返回我们的不透明类型
extern "C" fn __version() -> CStr = "version"

// 4. 封装成一个安全的 MoonBit 函数
fn version() -> String {
  __version().to_string()
}

C 侧 (在 ​cwrap.c​​ 中添加):

#include <string.h> // for strlen

// 这个函数负责将 char* 正确地转换为带 GC 头的 moonbit_string_t
moonbit_string_t cstr_to_moonbit_str(char *ptr) {
  if (ptr == NULL) {
    return moonbit_make_string(0, 0);
  }
  int32_t len = strlen(ptr);
  // moonbit_make_string 会分配一个带 GC 头的 MoonBit 字符串对象
  moonbit_string_t ms = moonbit_make_string(len, 0);
  for (int i = 0; i < len; i++) {
    ms[i] = (uint16_t)ptr[i]; // 假设是 ASCII 兼容的
  }
  // 注意:是否需要 free(ptr) 取决于 C 库的 API 约定。
  // 如果 version() 返回的内存需要调用者释放,这里就需要 free。
  return ms;
}

这个模式虽然初看有些繁琐,但它保证了内存安全,是处理 C 字符串返回值的标准做法。

3.3 指针的艺术:传递引用与数组 (The Art of Pointers: Passing by Reference and Arrays)​

C 语言大量使用指针来实现"输出参数"和传递数组。MoonBit 为此提供了专门的类型。

单个值的"输出"参数

当 C 函数使用指针来返回一个额外的值时,如 tan_with_errcode(double input, double* output)​,MoonBit 使用 Ref[T]​ 类型来对应。

extern "C" fn tan_with_errcode(input: Double, output: Ref[Double]) -> Int = "tan_with_errcode"

​Ref[T]​ 在 MoonBit 中是一个包含单个 T​ 类型字段的结构体。当它传递给 C 时,MoonBit 会传递这个结构体的地址。从 C 的角度看,一个指向 struct { T val; }​ 的指针和一个指向 T​ 的指针在内存地址上是等价的,因此可以直接工作。

数组:传递数据集合

当 C 函数需要处理一个数组时(例如 double* inputs​),MoonBit 使用 FixedArray[T]​ 类型来映射。FixedArray[T]​ 在内存中就是一块连续的 T​ 类型元素,其指针可以直接传递给 C。

extern "C" fn sin_array(len: Int, inputs: FixedArray[Double], outputs: FixedArray[Double]) -> Int = "sin_array"
extern "C" fn cos_array(len: Int, inputs: FixedArray[Double], outputs: FixedArray[Double]) -> Int = "cos_array"
extern "C" fn tan_array(len: Int, inputs: FixedArray[Double], outputs: FixedArray[Double]) -> Int = "tan_array"

3.4 外部类型:拥抱不透明的 C 结构体 (External Types: Embracing Opaque C Structs)​

对于 C 中的 struct​,比如 Complex​,最佳实践通常是将其视为一个"不透明类型"(Opaque Type)。我们只在 MoonBit 中创建一个对它的引用(或句柄),而不关心其内部的具体字段。

这通过 #extern type​ 语法实现:

#extern
type Complex

这个声明告诉 MoonBit:"存在一个名为 Complex​ 的外部类型。你不需要知道它的内部结构,只要把它当成一个指针大小的句柄来传递就行了。" 在生成的 C 代码中,Complex​ 类型会被处理成 void*​。这通常是安全的,因为所有对 Complex​ 的操作都是在 C 库内部完成的,MoonBit 侧只负责传递指针。

基于这个原则,我们可以为 mymath.h​ 中与 Complex​ 相关的函数编写 FFI:

// C: Complex* new_complex(double r, double i);
// 返回一个指向 Complex 的指针,在 MoonBit 中就是返回一个 Complex 句柄
extern "C" fn new_complex(r: Double, i: Double) -> Complex = "new_complex"

// C: void multiply(Complex* a, Complex* b, Complex** result);
// Complex* 对应 Complex,而 Complex** 对应 Ref[Complex]
extern "C" fn multiply(a: Complex, b: Complex, res: Ref[Complex]) -> Unit = "multiply"

// C: void init_n_complexes(int n, Complex** complex_array);
// Complex** 在这里作为数组使用,对应 FixedArray[Complex]
extern "C" fn init_n_complexes(n: Int, complex_array: FixedArray[Complex]) -> Unit = "init_n_complexes"

最佳实践:封装原生 FFI 直接暴露 FFI 函数会让使用者感到困惑(比如 Ref​ 和 FixedArray​)。强烈建议在 FFI 声明之上再构建一层对 MoonBit 用户更友好的 API。

// 在 Complex 类型上定义方法,隐藏 FFI 细节
fn Complex::mul(self: Complex, other: Complex) -> Complex {
  // 创建一个临时的 Ref 用于接收结果
  let res: Ref[Complex] = Ref::{ val: new_complex(0, 0) }
  multiply(self, other, res)
  res.val // 返回结果
}

fn init_n(n: Int) -> Array[Complex] {
  // 使用 FixedArray::make 创建数组
  let arr = FixedArray::make(n, new_complex(0, 0))
  init_n_complexes(n, arr)
  // 将 FixedArray 转换为对用户更友好的 Array
  Array::from_fixed_array(arr)
}

3.5 函数指针:当 C 需要回调 MoonBit (Function Pointers: When C Needs to Call Back)​

​mymath.h​ 中最复杂的函数是 for_each_complex​,它接受一个函数指针作为参数。

void for_each_complex(int n, Complex** arr, void (*call_back)(Complex*));

一个常见的误解是试图将 MoonBit 的闭包类型 (Complex) -> Unit​ 直接映射到 C 的函数指针。这是不行的,因为 MoonBit 的闭包在底层是一个包含两部分的结构体:一个指向实际函数代码的指针,以及一个指向其捕获的环境数据的指针。

为了传递一个纯粹的、无环境捕获的函数指针,MoonBit 提供了 FuncRef​ 类型:

extern "C" fn for_each_complex(
  n: Int,
  arr: FixedArray[Complex],
  call_back: FuncRef[(Complex) -> Unit] // 使用 FuncRef 包装函数类型
) -> Unit = "for_each_complex"

任何被 FuncRef​ 包裹的函数类型,在传递给 C 时,都会被转换成一个标准的 C 函数指针。

如何声明一个FuncRef​?只要使用let​就可以了,只要函数没有捕获外部变量,就可以声明成功。

fn print_complex(c: Complex) -> Unit { ... }

fn main {
  let print_complex : FuncRef[(Complex) -> Unit] = (c) => print_complex(c)
  // ...
}

第四站:高级课题——GC管理(Advanced Topic: GC Management)​

我们已经了解了大部分类型的转换问题,但还有一个非常重大的问题:内存管理。C 依赖手动的 malloc​/free​,而 MoonBit 拥有自动的垃圾回收(GC)。当 C 库创建了一个对象(如 new_complex​),谁来负责释放它?

可以不要GC吗? ​

一些库作者可能会选择不做GC,而是把所有的析构操作都留给用户。这种做法在一些库上有其合理性,因为有些库,例如一些高性能计算库,图形库等,为了提高性能或者稳定性,本身就会放弃掉一些GC特性,但带来的问题就是对程序员的水平要求较高。大多数库还是需要提供GC来增强用户体验的。

理想情况下,我们希望 MoonBit 的 GC 能够自动管理这些 C 对象的生命周期。MoonBit 提供了两种机制来实现这一点。

4.1 简单情况​

如果 C 结构体非常简单,并且你确信它的内存布局在所有平台上都是稳定不变的,你可以直接在 MoonBit 中重新定义它。

// mymath.h: typedef struct { double real; double img; } Complex;
// MoonBit:
struct Complex {
  r: Double,
  i: Double
}

这样做,Complex​ 就成了一个真正的 MoonBit 对象。MoonBit 编译器会自动为它管理内存,添加 GC 头。当你把它传递给 C 函数时,MoonBit 会传递一个指向其数据部分的指针,这通常是可行的。

但这种方法有很大的局限性:

  • 它要求你精确知道 C 结构体的内存布局、对齐方式等,这可能很脆弱。
  • 如果 C 函数返回一个 Complex*​,你不能直接使用它。你必须像处理字符串返回值一样,编写一个 C 包装函数,将 C 结构体的数据复制到一个新创建的、带 GC 头的 MoonBit Complex​ 对象中。

因此,这种方法只适用于最简单的情况。对于大多数场景,我们推荐更健壮的析构方案。

4.2 复杂情况,使用析构函数(Finalizer) (The Complex Situation: Using Finalizers)​

这是一种更通用和安全的方法。核心思想是:创建一个 MoonBit 对象来"包装"C 指针,并告诉 MoonBit 的 GC,当这个包装对象被回收时,应该调用一个特定的 C 函数(析构函数)来释放底层的 C 指针。

这个过程分为几步:

1. 在 MoonBit 中声明两种类型

#extern
type C_Complex // 代表原始的、不透明的 C 指针

type Complex C_Complex // 一个 MoonBit 类型,它内部包装了一个 C_Complex

​type Complex C_Complex​ 是一个特殊的声明,它创建了一个名为 Complex​ 的 MoonBit 对象类型,其内部有一个字段,类型为 C_Complex​。我们可以通过 .inner()​ 方法访问到这个内部字段。

2. 在 C 中提供析构函数和包装函数

我们需要一个 C 函数来释放 Complex​ 对象,以及一个函数来创建我们带 GC 功能的 MoonBit 包装对象。

C 侧 (在 ​cwrap.c​​ 中添加):

// mymath 库应该提供一个释放 Complex 的函数,假设是 free_complex
// void free_complex(Complex* c);

// 我们需要一个 void* 版本的析构函数给 MoonBit GC 使用
void free_complex_finalizer(void* obj) {
    // MoonBit 外部对象的布局是 { void (*finalizer)(void*); T data; }
    // 我们需要从 obj 中提取出真正的 Complex 指针
    // 假设 MoonBit 的 Complex 包装器只有一个字段
    Complex* c_obj = *((Complex**)obj);
    free_complex(c_obj); // 调用真正的析构函数, 如果mymath库提供的话
    // free(c_obj); // 如果是标准的 malloc 分配的
}

// 定义 MoonBit 的 Complex 包装器在 C 中的样子
typedef struct {
  Complex* val;
} MoonBit_Complex;

// 创建 MoonBit 包装对象的函数
MoonBit_Complex* new_mbt_complex(Complex* c_complex) {
  // `moonbit_make_external_obj` 是关键
  // 它创建一个由 GC 管理的外部对象,并注册其析构函数。
  MoonBit_Complex* mbt_complex = moonbit_make_external_obj(
      &free_complex_finalizer,
      sizeof(MoonBit_Complex)
  );
  mbt_complex->val = c_complex;
  return mbt_complex;
}

3. 在 MoonBit 中使用包装函数

现在,我们不直接调用 new_complex,而是调用我们的包装函数 new_mbt_complex。

// FFI 声明指向我们的 C 包装函数
extern "C" fn __new_managed_complex(c_complex: C_Complex) -> Complex = "new_mbt_complex"

// 原始的 C new_complex 函数返回一个裸指针
extern "C" fn __new_unmanaged_complex(r: Double, i: Double) -> C_Complex = "new_complex"

// 最终提供给用户的、安全的、GC 友好的 new 函数
fn Complex::new(r: Double, i: Double) -> Complex {
  let c_ptr = __new_unmanaged_complex(r, i)
  __new_managed_complex(c_ptr)
}

现在,当 Complex::new 创建的对象在 MoonBit 中不再被使用时,GC 会自动调用 free_complex_finalizer,从而安全地释放了 C 库分配的内存。

当需要将我们管理的 Complex 对象传递给其他 C 函数时,只需使用 .inner() 方法:

// 假设有一个C函数 `double length(Complex*);`
extern "C" fn length(c_complex: C_Complex) -> Double = "length"

fn Complex::length(self: Self) -> Double {
  // self.inner() 返回内部的 C_Complex (即 C 指针)
  length(self.inner())
}

结语 (Conclusion)​

这篇文章带你从基本类型,到复杂的结构体类型,再到函数指针类型,梳理了在MoonBit中做C-FFI的流程。末尾讨论了MoonBit管理c对象的GC问题。希望对广大读者的库开发有帮助。

Moonbit 与 llvm 共舞 下篇 - llvm后端生成

· 阅读需 18 分钟


引言​

在编程语言设计的过程中,语法前端负责理解和验证程序的结构与语义,而编译器后端则承担着将这些抽象概念转化为可执行机器代码的重任。后端的实现不仅需要对目标体系结构有深入的理解,更要掌握复杂的优化技术来生成高效的代码。

LLVM(Low Level Virtual Machine)作为现代编译器基础设施的集大成者,为我们提供了一个强大而灵活的解决方案。通过将程序转换为LLVM中间表示(Intermediate Representation, IR),我们可以利用LLVM成熟的工具链将代码编译到多种目标架构,包括RISC-V、ARM和x86等。

Moonbit的LLVM生态

Moonbit官方提供了两个重要的LLVM相关项目:

  • ​**llvm.mbt**​:原版LLVM的Moonbit语言绑定,提供对llvm-c接口的直接访问。需要安装完整的LLVM工具链,只能生成native后端,需要自行解决编译和链接的问题,但能够生成与原版LLVM完全兼容的IR。
  • ​**MoonLLVM**​:纯Moonbit实现的LLVM仿制版,无需外部依赖即可生成LLVM IR,支持JavaScript和WebAssembly后端

本文选择llvm.mbt​作为我们的工具,其API设计参考了Rust生态中广受好评的inkwell库。

在上篇《Moonbit 与 LLVM 共舞:实现现代编译器(上篇)》中,我们已经完成了从源代码到类型化抽象语法树的转换。本篇将承接这一成果,重点阐述代码生成的核心技术和实现细节。


第一章:LLVM类型系统的Moonbit表示​

在深入代码生成之前,我们需要首先理解llvm.mbt​如何在Moonbit的类型系统中表示LLVM的各种概念。LLVM的类型系统相当复杂,包含基本类型、复合类型和函数类型等多个层次。

Trait Object:类型的抽象表示​

在llvm.mbt​的API设计中,你会频繁遇到&Type​这一核心概念。这并非一个具体的struct或enum,而是一个Trait Object——可以将其理解为面向对象编程中抽象基类的函数式对等物。

// &Type是一个trait object,代表任意LLVM类型
let some_type: &Type = context.i32_type()

类型识别与转换​

要确定一个&Type​的具体类型,我们需要通过as_type_enum​接口进行运行时类型检查:

pub fn identify_type(ty: &Type) -> String {
  match ty.as_type_enum() {
    IntType(int_ty) => "Integer type with \{int_ty.get_bit_width()} bits"
    FloatType(float_ty) => "Floating point type"
    PointerType(ptr_ty) => "Pointer type"
    FunctionType(func_ty) => "Function type"
    ArrayType(array_ty) => "Array type"
    StructType(struct_ty) => "Structure type"
    VectorType(vec_ty) => "Vector type"
    ScalableVectorType(svec_ty) => "Scalable vector type"
    MetadataType(meta_ty) => "Metadata type"
  }
}

安全的类型转换策略​

当我们确信某个&Type​具有特定的类型时,有多种转换方式可供选择:

  1. 直接转换(适用于确定性场景)
let ty: &Type = context.i32_type()
let i32_ty = ty.into_int_type()  // 直接转换,错误由llvm.mbt处理
let bit_width = i32_ty.get_bit_width()  // 调用IntType特有的方法
  1. 防御性转换(推荐的生产环境做法)
let ty: &Type = get_some_type()  // 从某处获得的未知类型

guard ty.as_type_enum() is IntType(i32_ty) else {
  raise CodeGenError("Expected integer type, got \{ty}")
}

// 现在可以安全地使用i32_ty
let bit_width = i32_ty.get_bit_width()

复合类型的构造​

LLVM支持多种复合类型,这些类型通常通过基本类型的方法来构造:

pub fn create_composite_types(context: @llvm.Context) -> Unit {
  let i32_ty = context.i32_type()
  let f64_ty = context.f64_type()

  // 数组类型:[16 x i32]
  let i32_array_ty = i32_ty.array_type(16)

  // 函数类型:i32 (i32, i32)
  let add_func_ty = i32_ty.fn_type([i32_ty, i32_ty])

  // 结构体类型:{i32, f64}
  let struct_ty = context.struct_type([i32_ty, f64_ty])

  // 指针类型(LLVM 18+中所有指针都是opaque)
  let ptr_ty = i32_ty.ptr_type()

  // 输出类型信息用于验证
  println("Array type: \{i32_array_ty}")      // [16 x i32]
  println("Function type: \{add_func_ty}")    // i32 (i32, i32)
  println("Struct type: \{struct_ty}")        // {i32, f64}
  println("Pointer type: \{ptr_ty}")          // ptr
}

重要提醒:Opaque指针

自LLVM 18版本开始,所有指针类型都采用了opaque指针设计。这意味着无论指向什么类型,所有指针在IR中都表示为ptr​,指向的具体类型信息在类型系统中不再可见。


第二章:LLVM值系统与BasicValue概念​

相比类型系统,LLVM的值系统会复杂一些。llvm.mbt​与inkwell一致,将值分为两个重要的抽象层次。Value​ 和 BasicValue​。不同点在于在于区分值的创建来源和值的使用方式:

  • Value:关注值是如何产生的(常量、指令结果等)
  • BasicValue:关注值具有什么样的基本类型(整数、浮点数、指针等)

实际应用示例​

pub fn demonstrate_value_system(context: Context, builder: Builder) -> Unit {
  let i32_ty = context.i32_type()

  // 创建两个整数常量 - 这些直接就是IntValue
  let const1 = i32_ty.const_int(10)  // Value: IntValue, BasicValue: IntValue
  let const2 = i32_ty.const_int(20)  // Value: IntValue, BasicValue: IntValue

  // 执行加法运算 - 结果是一个指令InstructionValue
  let add_result = builder.build_int_add(const1, const2)

  // 在不同的上下文中,我们需要不同的视角:

  // 作为指令来检查其属性
  let instruction = add_result.as_instruction()
  println("Instruction opcode: \{instruction.get_opcode()}")

  // 作为基本值来获取其类型
  let basic_value = add_result.into_basic_value()
  println("Result type: \{basic_value.get_type()}")

  // 作为整数值来进行后续计算
  let int_value = add_result.into_int_value()
  let final_result = builder.build_int_mul(int_value, const1)
}

值类型的完整分类​

  1. ValueEnum:所有可能的值类型
pub enum ValueEnum {
  IntValue(IntValue)              // 整数值
  FloatValue(FloatValue)          // 浮点数值
  PointerValue(PointerValue)      // 指针值
  StructValue(StructValue)        // 结构体值
  FunctionValue(FunctionValue)    // 函数值
  ArrayValue(ArrayValue)          // 数组值
  VectorValue(VectorValue)        // 向量值
  PhiValue(PhiValue)             // Phi节点值
  ScalableVectorValue(ScalableVectorValue)  // 可伸缩向量值
  MetadataValue(MetadataValue)    // 元数据值
  CallSiteValue(CallSiteValue)    // 调用点值
  GlobalValue(GlobalValue)        // 全局值
  InstructionValue(InstructionValue)  // 指令值
} derive(Show)
  1. BasicValueEnum:具有基本类型的值
pub enum BasicValueEnum {
  ArrayValue(ArrayValue)              // 数组值
  IntValue(IntValue)                  // 整数值
  FloatValue(FloatValue)              // 浮点数值
  PointerValue(PointerValue)          // 指针值
  StructValue(StructValue)            // 结构体值
  VectorValue(VectorValue)            // 向量值
  ScalableVectorValue(ScalableVectorValue)  // 可伸缩向量值
} derive(Show)

💡 值转换的最佳实践​

在实际的代码生成过程中,我们经常需要在不同的值视角之间进行转换:

pub fn value_conversion_patterns(instruction_result: &Value) -> Unit {
  // 模式1:我知道这是什么类型,直接转换
  let int_val = instruction_result.into_int_value()

  // 模式2:我只需要一个基本值,不关心具体类型
  let basic_val = instruction_result.into_basic_value()

  // 模式3:防御性编程,检查后转换
  match instruction_result.as_value_enum() {
    // 处理整数值
    IntValue(int_val) => handle_integer(int_val)
    // 处理浮点值
    FloatValue(float_val) => handle_float(float_val)
    _ => raise CodeGenError("Unexpected value type")
  }
}

通过这种双层抽象,llvm.mbt​既保持了LLVM值系统的完整性,又为Moonbit开发者提供了直观易用的接口。


第三章:LLVM IR生成实战​

在理解了类型和值系统的基础上,让我们通过一个完整的示例来演示如何使用llvm.mbt​生成LLVM IR。这个示例将实现一个简单的 muladd​ 函数,展示从初始化到指令生成的完整流程。

基础设施初始化​

任何LLVM程序的开始都需要建立三个核心组件:

pub fn initialize_llvm() -> (Context, Module, Builder) {
  // 1. 创建LLVM上下文 - 所有LLVM对象的容器
  let context = @llvm.Context::create()

  // 2. 创建模块 - 函数和全局变量的容器
  let module = context.create_module("demo_module")

  // 3. 创建IR构建器 - 用于生成指令
  let builder = context.create_builder()

  (context, module, builder)
}

一个简单的函数生成示例​

让我们实现一个计算 (a * b) + c​ 的函数:

pub fn generate_muladd_function() -> String {
  // 初始化LLVM基础设施
  let (context, module, builder) = initialize_llvm()

  // 定义函数签名
  let i32_ty = context.i32_type()
  let func_type = i32_ty.fn_type([i32_ty, i32_ty, i32_ty])
  let func_value = module.add_function("muladd", func_type)

  // 创建函数入口基本块
  let entry_block = context.append_basic_block(func_value, "entry")
  builder.position_at_end(entry_block)

  // 获取函数参数
  let arg_a = func_value.get_nth_param(0).unwrap().into_int_value()
  let arg_b = func_value.get_nth_param(1).unwrap().into_int_value()
  let arg_c = func_value.get_nth_param(2).unwrap().into_int_value()

  // 生成计算指令
  let mul_result = builder.build_int_mul(arg_a, arg_b).into_int_value()
  let add_result = builder.build_int_add(mul_result, arg_c)

  // 生成返回指令
  let _ = builder.build_return(add_result)

  // 输出生成的IR
  module.dump()
}

生成的LLVM IR​

运行上述代码将产生以下LLVM中间表示:

; ModuleID = 'demo_module'
source_filename = "demo_module"

define i32 @muladd(i32 %0, i32 %1, i32 %2) {
entry:
  %3 = mul i32 %0, %1
  %4 = add i32 %3, %2
  ret i32 %4
}

💡 代码生成最佳实践​

  1. 命名约定

有返回值的指令,构建接口有一个name​的label argument,可以给指令的结果添加名称。

let mul_result = builder.build_int_mul(lhs, rhs, name="temp_product")
let final_result = builder.build_int_add(mul_result, offset, name="final_sum")
  1. 错误处理

使用raise而并非panic来进行错误处理,对不好直接确定的情况进行异常管理。

// 对可能失败的操作进行检查
match func_value.get_nth_param(index) {
  Some(param) => param.into_int_value()
  None => raise CodeGenError("Function parameter \{index} not found")
}

第四章:TinyMoonbit编译器实现​

现在让我们将注意力转向真正的编译器实现,将上篇文章中构建的抽象语法树转换为LLVM IR。

类型映射:从Parser到LLVM​

首先需要建立TinyMoonbit类型系统与LLVM类型系统之间的映射关系:

pub struct CodeGen {
  parser_program : Program                    // 源程序的AST表示
  llvm_context : @llvm.Context               // LLVM上下文
  llvm_module : @llvm.Module                 // LLVM模块
  builder : @llvm.Builder                    // IR构建器
  llvm_functions : Map[String, @llvm.FunctionValue]  // 函数映射表
}

pub fn convert_type(self : Self, parser_type : Type) -> &@llvm.Type raise {
  match parser_type {
    Type::Unit => self.llvm_context.void_type() as &@llvm.Type
    Type::Bool => self.llvm_context.bool_type()
    Type::Int => self.llvm_context.i32_type()
    Type::Double => self.llvm_context.f64_type()
    // 可以根据需要扩展更多类型
  }
}

环境管理:变量到值的映射​

在代码生成阶段,我们需要维护一个从变量名到LLVM值的映射关系:

pub struct Env {
  parent : Env?                        // 父环境引用
  symbols : Map[String, &@llvm.Value]        // 局部变量映射

  // 全局信息
  codegen : CodeGen                           // 代码生成器引用
  parser_function : Function                  // 当前函数的AST
  llvm_function : @llvm.FunctionValue         // 当前函数的LLVM表示
}

pub fn get_symbol(self : Self, name : String) -> &@llvm.Value? {
  match self.symbols.get(name) {
    Some(value) => Some(value)
    None =>
      match self.parent {
        Some(parent_env) => parent_env.get_symbol(name)
        None => None
      }
  }
}

变量处理:内存分配策略​

TinyMoonbit作为一个系统级语言,支持变量的重新赋值。在LLVM IR的SSA(Static Single Assignment)形式中,我们需要采用alloca + load/store的模式来实现可变变量:

pub fn Stmt::emit(self : Self, env : Env) -> Unit raise {
  match self {
    // 变量声明:例如let x : Int = 5;
    Let(var_name, var_type, init_expr) => {
      // 转换类型并分配栈空间
      let llvm_type = env.codegen.convert_type(var_type)
      let alloca = env.codegen.builder.build_alloca(llvm_type, var_name)

      // 将分配的指针记录到符号表
      env.symbols.set(var_name, alloca as &@llvm.Value)

      // 计算初始化表达式的值
      let init_value = init_expr.emit(env).into_basic_value()

      // 将初始值存储到分配的内存
      let _ = env.codegen.builder.build_store(alloca, init_value)
    }

    // 变量赋值:x = 10;
    Assign(var_name, rhs_expr) => {
      // 从符号表获取变量的内存地址
      guard let Some(var_ptr) = env.get_symbol(var_name) else {
        raise CodeGenError("Undefined variable: \{var_name}")
      }

      // 计算右侧表达式的值
      let rhs_value = rhs_expr.emit(env).into_basic_value()

      // 存储新值到变量内存
      let _ = env.codegen.builder.build_store(var_ptr, rhs_value)
    }

    // 其他语句类型...
    _ => { /* 处理其他语句 */ }
  }
}

设计决策:为什么使用alloca?

在函数式语言中,不可变变量可以直接映射为SSA值。但TinyMoonbit支持变量重新赋值,这与SSA的"每个变量只赋值一次"原则冲突。

alloca + load/store 模式是处理可变变量的标准做法:

  • ​alloca​:在栈上分配内存空间
  • ​store​:将值写入内存
  • ​load​:从内存读取值

LLVM的优化过程会自动将简单的alloca转换回值形式(mem2reg优化)。

表达式代码生成​

表达式的代码生成相对直观,主要是根据表达式类型调用相应的指令构建方法:

fn Expr::emit(self: Self, env: Env) -> &@llvm.Value raise {
  match self {
    AtomExpr(atom_expr, ..) => atom_expr.emit(env)
    Unary("-", expr, ty = Some(Int)) => {
      let value = expr.emit().into_int_value()
      let zero = env.gen.llvm_ctx.i32_type().const_zeor()
      env.gen.builder.build_int_sub(zero, value)
    }
    Unary("-", expr, ty = Some(Double)) => {
      let value = expr.emit().into_float_value()
      env.gen.builder.build_float_neg(value)
    }
    Binary("+", lhs, rhs, ty=Some(Int)) => {
      let lhs_val = lhs.emit().into_int_value()
      let rhs_val = rhs.emit().into_int_value()
      env.gen.builder.build_int_add(lhs_val, rhs_val)
    }
    // ... others
  }
}

技术细节:浮点数取负

注意在处理浮点数取负时,我们使用 build_float_neg​ 而不是用零减去操作数。这是因为:

  1. IEEE 754标准:浮点数有特殊值(如NaN、∞),简单的减法可能产生不正确的结果
  2. 性能考虑:专用的否定指令在现代处理器上通常更高效
  3. 精度保证:避免了不必要的舍入误差

第五章:控制流指令的实现​

控制流是程序逻辑的骨架,包括条件分支和循环结构。在LLVM IR中,控制流通过基本块(Basic Blocks)和分支指令来实现。每个基本块代表一个没有内部跳转的指令序列,块与块之间通过分支指令连接。

条件分支:if-else语句的实现​

条件分支需要创建多个基本块来表示不同的执行路径:

fn Stmt::emit(self: Self, env: Env) -> Unit raise {
  let ctx = env.gen.llvm_ctx
  let func = env.llvm_func
  let builder = env.gen.builder
  match self {
    If(cond, then_stmts, else_stmts) => {
      let cond_val = cond.emit(env).into_int_value()

      // 创建三个基本块
      let then_block = ctx.append_basic_block(llvm_func)
      let else_block = ctx.append_basic_block(llvm_func)
      let merge_block = ctx.append_basic_block(llvm_func)

      // 创建跳转指令
      let _ = builder.build_conditional_branch(
        cond_val, then_block, else_block,
      )

      // 生成then_block的代码
      builder.position_at_end(then_block)
      let then_env = self.subenv()
      then_stmts.each(s => s.emitStmt(then_env))
      let _ = builder.build_unconditional_branch(merge_block)

      // 生成else_block的代码
      builder.position_at_end(else_block)
      let else_env = self.subenv()
      else_stmts.each(s => s.emitStmt(else_env))
      let _ = builder.build_unconditional_branch(merge_block)

      // 代码生成完毕后,builder的位置要在merge_block上
      builder.position_at_end(merge_block)

    }
    // ...
  }
}

生成的LLVM IR示例​

对于以下TinyMoonbit代码:

if x > 0 {
  y = x + 1;
} else {
  y = x - 1;
}

将生成类似这样的LLVM IR:

  %1 = load i32, ptr %x, align 4
  %2 = icmp sgt i32 %1, 0
  br i1 %2, label %if.then, label %if.else

if.then:                                          ; preds = %0
  %3 = load i32, ptr %x, align 4
  %4 = add i32 %3, 1
  store i32 %4, ptr %y, align 4
  br label %if.end

if.else:                                          ; preds = %0
  %5 = load i32, ptr %x, align 4
  %6 = sub i32 %5, 1
  store i32 %6, ptr %y, align 4
  br label %if.end

if.end:                                           ; preds = %if.else, %if.then
  ; 后续代码...

循环结构:while语句的实现​

循环的实现需要特别注意条件检查和循环体的正确连接:

fn Stmt::emit(self: Self, env: Env) -> Unit raise {
  let ctx = env.gen.llvm_ctx
  let func = env.llvm_func
  let builder = env.gen.builder
  match self {
    While(cond, body) => {
      // 生成三个块
      let cond_block = ctx.append_basic_block(.llvm_func)
      let body_block = ctx.append_basic_block(llvm_func)
      let merge_block = ctx.append_basic_block(llvm_func)

      // 首先无条件跳转到cond块
      let _ = builder.build_unconditional_branch(cond_block)
      builder.position_at_end(cond_block)

      // 在cond块内生成代码,以及条件跳转指令
      let cond_val = cond.emit().into_int_value()
      let _ = builder.build_conditional_branch(
        cond_val, body_block, merge_block,
      )
      builder.position_at_end(body_block)

      // 对body块生成代码,末尾需要一个无条件跳转指令,到cond块
      let body_env = self.subenv()
      body.each(s => s.emitStmt(body_env))
      let _ = builder.build_unconditional_branch(cond_block)

      // 代码生成结束以后,跳转到merge block
      builder.position_at_end(merge_block)
    }
    // ...
  }
}

生成的LLVM IR示例​

对于TinyMoonbit代码:

while i < 10 {
  i = i + 1;
}

将生成:

  br label %while.cond

while.cond:                                       ; preds = %while.body, %0
  %1 = load i32, ptr %i, align 4
  %2 = icmp slt i32 %1, 10
  br i1 %2, label %while.body, label %while.end

while.body:                                       ; preds = %while.cond
  %3 = load i32, ptr %i, align 4
  %4 = add i32 %3, 1
  store i32 %4, ptr %i, align 4
  br label %while.cond

while.end:                                        ; preds = %while.cond
  ; 后续代码...

**💡 控制流设计要点 **

  1. 基本块的命名策略

​append_basic_block​ 函数同样有name​这个label argument。

// 使用描述性的块名称,便于调试和理解
let then_block = context.append_basic_block(func, name="if.then")
let else_block = context.append_basic_block(func, name="if.else")
let merge_block = context.append_basic_block(func, name="if.end")
  1. 作用域管理
// 为每个分支和循环体创建独立的作用域
let branch_env = env.sub_env()
branch_stmts.each( stmt => stmt.emit(branch_env) }
  1. 构建器位置管理

末尾注意将指令构建器放到正确的基本块上。

// 始终确保构建器指向正确的基本块
builder.position_at_end(merge_block)
// 在这个块中生成指令...

第六章:从LLVM IR到机器代码​

在生成完整的LLVM IR之后,我们需要将其转换为目标机器的汇编代码。虽然llvm.mbt​提供了完整的目标机器配置API,但对于学习目的,我们可以使用更简便的方法。

使用llc工具链进行编译​

最直接的方法是将生成的LLVM IR输出到文件,然后使用LLVM工具链进行编译:

调用Module​的dump​函数即可,也可以使用println​函数。

let gen : CodeGen = ...
let prog = gen.llvm_prog
prog.dump() // 更建议使用dump,会比println快一点,效果相同

// or println(prog)

完整的编译流程示例​

让我们看一个完整的从源代码到汇编代码的编译流程:

  1. TinyMoonbit源代码
fn factorial(n: Int) -> Int {
  if n <= 1 {
    return 1;
  }
  return n * factorial(n - 1);
}

fn main() -> Unit {
  let result: Int = factorial(5);
  print_int(result);
}
  1. 生成的LLVM IR
; ModuleID = 'tinymoonbit'
source_filename = "tinymoonbit"

define i32 @factorial(i32 %0) {
entry:
  %1 = alloca i32, align 4
  store i32 %0, ptr %1, align 4
  %2 = load i32, ptr %1, align 4
  %3 = icmp sle i32 %2, 1
  br i1 %3, label %4, label %6

4:                                                ; preds = %entry
  ret i32 1

6:                                                ; preds = %entry
  %7 = load i32, ptr %1, align 4
  %8 = load i32, ptr %1, align 4
  %9 = sub i32 %8, 1
  %10 = call i32 @factorial(i32 %9)
  %11 = mul i32 %7, %10
  ret i32 %11
}

define void @main() {
entry:
  %0 = alloca i32, align 4
  %1 = call i32 @factorial(i32 5)
  store i32 %1, ptr %0, align 4
  %2 = load i32, ptr %0, align 4
  call void @print_int(i32 %2)
  ret void
}

declare void @print_int(i32 %0)
  1. 使用LLC生成RISC-V汇编
# 生成llvm ir
moon run main --target native > fact.ll

# 生成RISC-V 64位汇编代码
llc -march=riscv64 -mattr=+m -o fact.s fact.ll
  1. 生成的RISC-V汇编片段
factorial:
.Lfunc_begin0:
	.cfi_startproc
	addi	sp, sp, -32
	.cfi_def_cfa_offset 32
	sd	ra, 24(sp)
	.cfi_offset ra, -8
	sd	s0, 16(sp)
	.cfi_offset s0, -16
	addi	s0, sp, 32
	.cfi_def_cfa s0, 0
	sw	a0, -20(s0)
	lw	a0, -20(s0)
	li	a1, 1
	blt	a1, a0, .LBB0_2
	li	a0, 1
	j	.LBB0_3
.LBB0_2:
	lw	a0, -20(s0)
	lw	a1, -20(s0)
	addi	a1, a1, -1
	sw	a0, -24(s0)
	mv	a0, a1
	call	factorial
	lw	a1, -24(s0)
	mul	a0, a1, a0
.LBB0_3:
	ld	ra, 24(sp)
	ld	s0, 16(sp)
	addi	sp, sp, 32
	ret

结语​

通过本系列的两篇文章,我们完成了一个功能完整的编译器实现。尽管功能简单,但比较完整。从字符流的词法分析,到抽象语法树的构建,再到LLVM IR的生成和机器代码的输出。

回顾​

上篇:

  • 基于模式匹配的优雅词法分析器
  • 递归下降语法分析器的实现
  • 完整的类型检查系统
  • 环境链作用域管理

下篇:

  • LLVM类型和值系统的深入理解
  • SSA形式下的变量管理策略
  • 控制流指令的正确实现
  • 完整的代码生成流水线

Moonbit在编译器开发中的优势​

通过这个实践项目,我们深刻体会到了Moonbit在编译器构建领域的独特价值:

  1. 表达力强大的模式匹配:极大简化了AST处理和类型分析的复杂度。
  2. 函数式编程范式:不可变数据结构和纯函数使得编译器逻辑更加清晰可靠。
  3. 现代化的类型系统:trait对象、泛型和错误处理机制提供了充分的抽象能力。
  4. 优秀的工程特性:derive功能、JSON序列化等特性显著提升了开发效率。

结语​

编译器技术代表了计算机科学理论与工程实践的完美结合。通过Moonbit这一现代化的工具,我们能够以更加优雅和高效的方式探索这个古老而又充满活力的领域。

希望本系列文章能够为读者在编译器设计的道路上提供一个有力的帮助。

学习资源推荐


Moonbit 与 llvm 共舞 上篇 - 实现语法前端

· 阅读需 17 分钟


引言​

编程语言设计与编译器实现历来被视为计算机科学领域中最具挑战性的课题之一。传统的编译器教学路径往往要求学生首先掌握复杂的理论基础:

  • 自动机理论:有限状态自动机与正则表达式
  • 类型理论:λ演算与类型系统的数学基础
  • 计算机体系结构:从汇编语言到机器码的底层实现

然而,Moonbit作为一门专为现代开发环境设计的函数式编程语言,为我们提供了一个全新的视角。它不仅具备严谨的类型系统和卓越的内存安全保障,更重要的是,其丰富的语法特性和为AI时代量身定制的工具链,使得Moonbit成为学习和实现编译器的理想选择。

系列概述 本系列文章将通过构建一个名为TinyMoonbit的小型编程语言编译器,深入探讨现代编译器实现的核心概念和最佳实践。

  • 上篇:聚焦语言前端的实现,包括词法分析、语法解析和类型检查,最终生成带有完整类型标记的抽象语法树
  • 下篇:深入代码生成阶段,利用Moonbit官方的llvm.mbt​绑定库,将语法树转换为LLVM中间表示,并最终生成RISC-V汇编代码

TinyMoonbit 语言设计​

TinyMoonbit是一种系统级编程语言,其抽象层次与C语言相当。虽然在语法设计上大量借鉴了Moonbit的特性,但TinyMoonbit实际并非Moonbit语言的子集,而是一个为测试llvm.mbt​功能完备性兼具教学作用的简化版本。

注:由于篇幅限制,本系列文章所提到的TinyMoonbit实现比真正的TinyMoonbit更加简单,完整版本请参考 TinyMoonbitLLVM。

核心特性​

TinyMoonbit提供了现代系统编程所需的基础功能:

  • ✅ 底层内存操作:直接的指针操作和内存管理
  • ✅ 控制流结构:条件分支、循环和函数调用
  • ✅ 类型安全:静态类型检查和明确的类型声明
  • ❌ 简化设计:为降低实现复杂度,不支持类型推导和闭包等高级特性

语法示例​

让我们通过一个经典的斐波那契数列实现来展示TinyMoonbit的语法:

extern fn print_int(x : Int) -> Unit;

// 递归实现斐波那契数列
fn fib(n : Int) -> Int {
  if n <= 1 {
    return n;
  }
  return fib(n - 1) + fib(n - 2);
}

fn main() -> Unit {
  print_int(fib(10));
}

编译目标​

经过完整的编译流程后,上述代码将生成如下的LLVM中间表示:

; ModuleID = 'tinymoonbit'
source_filename = "tinymoonbit"

define i32 @fib(i32 %0) {
entry:
  %1 = alloca i32, align 4
  store i32 %0, ptr %1, align 4
  %2 = load i32, ptr %1, align 4
  %3 = icmp sle i32 %2, 1
  br i1 %3, label %4, label %6

4:                                                ; preds = %entry
  %5 = load i32, ptr %1, align 4
  ret i32 %5

6:                                                ; preds = %4, %entry
  %7 = load i32, ptr %1, align 4
  %8 = sub i32 %7, 1
  %9 = call i32 @fib(i32 %8)
  %10 = load i32, ptr %1, align 4
  %11 = sub i32 %10, 2
  %12 = call i32 @fib(i32 %11)
  %13 = add i32 %9, %12
  ret i32 %13
}

define void @main() {
entry:
  %0 = call i32 @fib(i32 10)
  call void @print_int(i32 %0)
}

declare void @print_int(i32 %0)

第二章:词法分析​

词法分析(Lexical Analysis)构成了编译过程的第一道关卡,其核心使命是将连续的字符流转换为具有语义意义的词法单元(Tokens)序列。这个看似简单的转换过程,实际上是整个编译器流水线的基石。

从字符到符号:Token的设计与实现​

考虑以下代码片段:

let x : Int = 5;

经过词法分析器处理后,将产生如下的Token序列:

(Keyword "let") → (Identifier "x") → (Symbol ":") →
(Type "Int") → (Operator "=") → (IntLiteral 5) → (Symbol ";")

这个转换过程需要处理多种复杂情况:

  1. 空白符过滤:跳过空格、制表符和换行符
  2. 关键字识别:区分保留字与用户定义标识符
  3. 数值解析:正确识别整数、浮点数的边界
  4. 运算符处理:区分单字符和多字符运算符

Token类型系统设计​

基于TinyMoonbit的语法规范,我们将所有可能的符号分类为以下Token类型:

pub enum Token {
  Bool(Bool)       // 布尔值:true, false
  Int(Int)         // 整数:1, 2, 3, ...
  Double(Double)   // 浮点数:1.0, 2.5, 3.14, ...
  Keyword(String)  // 保留字:let, if, while, fn, return
  Upper(String)    // 类型标识符:首字母大写,如 Int, Double, Bool
  Lower(String)    // 变量标识符:首字母小写,如 x, y, result
  Symbol(String)   // 运算符和标点:+, -, *, :, ;, ->
  Bracket(Char)    // 括号类:(, ), [, ], {, }
  EOF              // 文件结束标记
} derive(Show, Eq)

利用模式匹配​

Moonbit的强大模式匹配能力使我们能够以一种前所未有的优雅方式实现词法分析器。与传统的有限状态自动机方法相比,这种基于模式匹配的实现更加直观和易于理解。

核心分析函数​

pub fn lex(code: String) -> Array[Token] {
  let tokens = Array::new()

  loop code[:] {
    // 跳过空白字符
    [' ' | '\n' | '\r' | '\t', ..rest] =>
      continue rest

    // 处理单行注释
    [.."//", ..rest] =>
      continue loop rest {
        ['\n' | '\r', ..rest_str] => break rest_str
        [_, ..rest_str] => continue rest_str
        [] as rest_str => break rest_str
      }

    // 识别多字符运算符(顺序很重要!)
    [.."->", ..rest] => { tokens.push(Symbol("->")); continue rest }
    [.."==", ..rest] => { tokens.push(Symbol("==")); continue rest }
    [.."!=", ..rest] => { tokens.push(Symbol("!=")); continue rest }
    [.."<=", ..rest] => { tokens.push(Symbol("<=")); continue rest }
    [..">=", ..rest] => { tokens.push(Symbol(">=")); continue rest }

    // 识别单字符运算符和标点符号
    [':' | '.' | ',' | ';' | '+' | '-' | '*' |
     '/' | '%' | '>' | '<' | '=' as c, ..rest] => {
      tokens.push(Symbol("\{c}"))
      continue rest
    }

    // 识别括号
    ['(' | ')' | '[' | ']' | '{' | '}' as c, ..rest] => {
      tokens.push(Bracket(c))
      continue rest
    }

    // 识别标识符和字面量
    ['a'..='z', ..] as code => {
      let (tok, rest) = lex_ident(code);
      tokens.push(tok)
      continue rest
    }

    ['A'..='Z', ..] => { ... }
    ['0'..='9', ..] => { ... }

    // 到达文件末尾
    [] => { tokens.push(EOF); break tokens }
  }
}

关键字识别策略​

标识符解析需要特别处理关键字的识别:

pub fn let_ident(rest: @string.View) -> (Token, @string.View) {
  // 预定义关键字映射表
  let keyword_map = Map.from_array([
    ("let", Token::Keyword("let")),
    ("fn", Token::Keyword("fn")),
    ("if", Token::Keyword("if")),
    ("else", Token::Keyword("else")),
    ("while", Token::Keyword("while")),
    ("return", Token::Keyword("return")),
    ("extern", Token::Keyword("extern")),
    ("true", Token::Bool(true)),
    ("false", Token::Bool(false)),
  ])

  let identifier_chars = Array::new()
  let remaining = loop rest {
    ['a'..='z' | 'A'..='Z' | '0'..='9' | '_' as c, ..rest_str] => {
      identifier_chars.push(c)
      continue rest_str
    }
    _ as rest_str => break rest_str
  }

  let ident = String::from_array(identifier_chars)
  let token = keyword_map.get(identifier).or_else(() => Token::Lower(ident))

  (token, remaining)
}

💡 Moonbit语法特性深度解析​

上述词法分析器的实现充分展示了Moonbit在编译器开发中的几个突出优势:

  1. 函数式循环构造
loop initial_value {
  pattern1 => continue new_value1
  pattern2 => continue new_value2
  pattern3 => break final_value
}

​loop​并非传统意义上的循环结构,而是一种函数式循环:

  • 接受一个初始参数作为循环状态
  • 通过模式匹配处理不同情况
  • ​continue​传递新状态到下一次迭代
  • ​break​终止循环并返回最终值
  1. 字符串视图与模式匹配

Moonbit的字符串模式匹配功能极大简化了文本处理:

// 匹配单个字符
['a', ..rest] => // 以字符'a'开头

// 匹配字符范围
['a'..='z' as c, ..rest] => // 小写字母,绑定到变量c

// 匹配字符串字面量
[.."hello", ..rest] => // 等价于 ['h','e','l','l','o', ..rest]

// 匹配多个可能的字符
[' ' | '\t' | '\n', ..rest] => // 任意空白字符
  1. 模式匹配优先级的重要性

⚠️ 重要提醒:匹配顺序至关重要

在编写模式匹配规则时,必须将更具体的模式放在更一般的模式之前。例如:

// ✅ 正确的顺序
loop code[:] {
  [.."->", ..rest] => { ... }     // 先匹配多字符运算符
  ['-' | '>' as c, ..rest] => { ... }  // 再匹配单字符
}

// ❌ 错误的顺序 - "->"将永远无法被匹配
loop code[:] {
  ['-' | '>' as c, ..rest] => { ... }
  [.."->", ..rest] => { ... }     // 永远不会执行
}

通过这种基于模式匹配的方法,我们不仅避免了复杂的状态机实现,还获得了更清晰、更容易维护的代码结构。


第三章:语法分析与抽象语法树构建​

语法分析(Syntactic Analysis)是编译器的第二个核心阶段,其任务是将词法分析产生的Token序列重新组织为具有层次结构的抽象语法树(Abstract Syntax Tree, AST)。这个过程不仅要验证程序是否符合语言的语法规则,更要为后续的语义分析和代码生成提供结构化的数据表示。

抽象语法树设计:程序的结构化表示​

在构建语法分析器之前,我们需要精心设计AST的结构。这个设计决定了如何表示程序的语法结构,以及后续编译阶段如何处理这些结构。

1. 核心类型系统​

首先,我们定义TinyMoonbit类型系统在AST中的表示:

pub enum Type {
  Unit    // 单位类型,表示无返回值
  Bool    // 布尔类型:true, false
  Int     // 32位有符号整数
  Double  // 64位双精度浮点数
} derive(Show, Eq, ToJson)

pub fn parse_type(type_name: String) -> Type {
  match type_name {
    "Unit" => Type::Unit
    "Bool" => Type::Bool
    "Int" => Type::Int
    "Double" => Type::Double
    _ => abort("Unknown type: \{type_name}")
  }
}

2. 分层的AST节点设计​

我们采用分层设计来清晰地表示程序的不同抽象层次:

  1. 原子表达式(AtomExpr) 代表不可再分解的基本表达式单元:
pub enum AtomExpr {
  Bool(Bool)                                    // 布尔字面量
  Int(Int)                                      // 整数字面量
  Double(Double)                                // 浮点数字面量
  Var(String, mut ty~ : Type?)                  // 变量引用
  Paren(Expr, mut ty~ : Type?)                  // 括号表达式
  Call(String, Array[Expr], mut ty~ : Type?)    // 函数调用
} derive(Show, Eq, ToJson)
  1. 复合表达式(Expr) 可以包含运算符和多个子表达式的更复杂结构:
pub enum Expr {
  AtomExpr(AtomExpr, mut ty~ : Type?)          // 原子表达式包装
  Unary(String, Expr, mut ty~ : Type?)         // 一元运算:-, !
  Binary(String, Expr, Expr, mut ty~ : Type?)  // 二元运算:+, -, *, /, ==, !=, 等
} derive(Show, Eq, ToJson)
  1. 语句(Stmt) 代表程序中的可执行单元:
pub enum Stmt {
  Let(String, Type, Expr)                      // 变量声明:let x : Int = 5;
  Assign(String, Expr)                         // 赋值语句:x = 10;
  If(Expr, Array[Stmt], Array[Stmt])           // 条件分支:if-else
  While(Expr, Array[Stmt])                     // 循环语句:while
  Return(Expr?)                                // 返回语句:return expr;
  Expr(Expr)                                   // 单表达式语句
} derive(Show, Eq, ToJson)
  1. 顶层结构 函数定义和完整程序:
pub struct Function {
  name : String                     // 函数名
  params : Array[(String, Type)]    // 参数列表:[(参数名, 类型)]
  ret_ty : Type                     // 返回类型
  body : Array[Stmt]                // 函数体语句序列
} derive(Show, Eq, ToJson)

// 程序定义为函数名到函数定义的映射
pub type Program Map[String, Function]

设计要点:类型标记的可变性

注意到每个表达式节点都包含一个 mut ty~ : Type?​ 字段。这个设计允许我们在类型检查阶段填充类型信息,而不需要重新构建整个AST。

递归下降解析:自顶向下的构建策略​

递归下降(Recursive Descent)是一种自顶向下的语法分析方法,其核心思想是为每个语法规则编写一个对应的解析函数。在Moonbit中,模式匹配使这种方法的实现变得异常优雅。

解析原子表达式​

pub fn parse_atom_expr(
  tokens: ArrayView[Token]
) -> (AtomExpr, ArrayView[Token]) raise {
  match tokens {
    // 解析字面量
    [Bool(b), ..rest] => (AtomExpr::Bool(b), rest)
    [Int(i), ..rest] => (AtomExpr::Int(i), rest)
    [Double(d), ..rest] => (AtomExpr::Double(d), rest)

    // 解析函数调用:func_name(arg1, arg2, ...)
    [Lower(func_name), Bracket('('), ..rest] => {
      let (args, rest) = parse_argument_list(rest)
      match rest {
        [Bracket(')'), ..remaining] =>
          (AtomExpr::Call(func_name, args, ty=None), remaining)
        _ => raise SyntaxError("Expected ')' after function arguments")
      }
    }

    // 解析变量引用
    [Lower(var_name), ..rest] =>
      (AtomExpr::Var(var_name, ty=None), rest)

    // 解析括号表达式:(expression)
    [Bracket('('), ..rest] => {
      let (expr, rest) = parse_expression(rest)
      match rest {
        [Bracket(')'), ..remaining] =>
          (AtomExpr::Paren(expr, ty=None), remaining)
        _ => raise SyntaxError("Expected ')' after expression")
      }
    }

    _ => raise SyntaxError("Expected atomic expression")
  }
}

解析语句​

语句解析需要根据开头的关键字分发到不同的处理函数:

pub fn parse_stmt(tokens : ArrayView[Token]) -> (Stmt, ArrayView[Token]) {
  match tokens {
    // 解析let语句
    [Keyword("let"), Lower(var_name), Symbol(":"), ..] => { /* ... */ }

    // 解析if/while/return语句
    [Keyword("if"), .. rest] => parse_if_stmt(rest)
    [Keyword("while"), .. rest] => parse_while_stmt(rest)
    [Keyword("return"), .. rest] => { /* ... */ }

    // 解析赋值语句
    [Lower(_), Symbol("="), .. rest] => parse_assign_stmt(tokens)

    // 解析单表达式语句
    [Lower(_), Symbol("="), .. rest] => parse_single_expr_stmt(tokens)

    _ => { /* 错误处理 */ }
  }
}

难点:处理运算符优先级:

表达式解析中最复杂的部分是处理运算符优先级,我们需要确保1 + 2 _ 3被正确解析为1 + (2 _ 3)而不是(1 + 2) * 3。

💡 Moonbit高级特性应用​

自动派生功能​

pub enum Expr {
  // ...
} derive(Show, Eq, ToJson)

Moonbit的 derive​ 功能自动为类型生成常用的实现,这里我们使用三个:

  • ​​Show​​:提供调试输出功能
  • ​​Eq​​:支持相等性比较
  • ​​ToJson​​:序列化为JSON格式,便于调试和持久化

这些自动生成的功能在编译器开发中极为有用,特别是在调试和测试阶段。

错误处理机制​

pub fn parse_expression(tokens: ArrayView[Token]) -> (Expr, ArrayView[Token]) raise {
  // raise关键字表示此函数可能抛出异常
}

Moonbit的 raise​ 机制提供了结构化的错误处理,使得语法错误能够被准确定位和报告。

通过这种分层设计和递归下降的解析策略,我们构建了一个既灵活又高效的语法分析器,为后续的类型检查阶段奠定了坚实的基础。


第四章:类型检查与语义分析​

语义分析是编译器设计中承上启下的关键阶段。虽然语法分析确保了程序结构的正确性,但这并不意味着程序在语义上是有效的。类型检查作为语义分析的核心组成部分,负责验证程序中所有操作的类型一致性,确保类型安全和运行时的正确性。

作用域管理:构建环境链​

类型检查面临的首要挑战是正确处理变量的作用域(Scope)。在程序的不同层次(全局、函数、块级别),同一个变量名可能指向不同的实体。我们采用环境链(Environment Chain)的经典设计来解决这个问题:

pub struct TypeEnv[K, V] {
  parent : TypeEnv[K, V]?     // 指向父环境的引用
  data : Map[K, V]            // 当前环境的变量绑定
}

环境链的核心是变量查找算法,它遵循词法作用域的规则:

pub fn TypeEnv::get[K : Eq + Hash, V](self : Self[K, V], key : K) -> V? {
  match self.data.get(key) {
    Some(value) => Some(value)    // 在当前环境中找到
    None =>
      match self.parent {
        Some(parent_env) => parent_env.get(key)  // 递归查找父环境
        None => None              // 到达顶层环境,变量未定义
      }
  }
}

设计原则:词法作用域

这种设计确保了变量的查找遵循词法作用域规则:

  1. 首先在当前作用域中查找
  2. 如果未找到,向上层作用域递归查找
  3. 直到找到变量或到达全局作用域

类型检查器架构​

单纯的环境管理还不足以完成类型检查任务。某些操作(如函数调用)需要访问全局的程序信息。因此,我们设计了一个综合的类型检查器:

pub struct TypeChecker {
  local_env : TypeEnv[String, Type]    // 本地变量环境
  current_func : Function              // 当前检查的函数
  program : Program                    // 完整的程序信息
}

部分节点类型检查的实现​

类型检查器的核心是对不同AST节点应用相应的类型规则。以下是表达式类型检查的实现:

pub fn Expr::check_type(
  self : Self,
  env : TypeEnv[String, Type]
) -> Type raise {
  match self {
    // 原子表达式的类型检查
    AtomExpr(atom_expr, ..) as node => {
      let ty = atom_expr.check_type(env)
      node.ty = Some(ty)  // 填充类型信息
      ty
    }

    // 一元运算的类型检查
    Unary("-", expr, ..) as node => {
      let ty = expr.check_type(env)
      node.ty = Some(ty)
      ty
    }

    // 二元运算的类型检查
    Binary(""+, lhs, rhs, ..) as node => {
      let lhs_type = lhs.check_type(env)
      let rhs_type = rhs.check_type(env)

      // 确保操作数类型一致
      guard lhs_type == rhs_type else {
        raise TypeCheckError(
          "Binary operation requires matching types, got \{lhs_type} and \{rhs_type}"
        )
      }

      let result_type = match op {
        // 比较运算符总是返回布尔值
        "==" | "!=" | "<" | "<=" | ">" | ">=" => Type::Bool

        // 算术运算符等保持操作数类型
        _ => lhs_type
      }

      node.ty = Some(result_type)
      result_type
    }
  }
}

** 💡 Moonbit枚举修改技巧 **

在类型检查过程中,我们需要为AST节点填充类型信息。Moonbit提供了一种优雅的方式来修改枚举变体的可变字段:

pub enum Expr {
  AtomExpr(AtomExpr, mut ty~ : Type?)
  Unary(String, Expr, mut ty~ : Type?)
  Binary(String, Expr, Expr, mut ty~ : Type?)
} derive(Show, Eq, ToJson)

通过在模式匹配中使用 as​ 绑定,我们可以获得对枚举变体的引用并修改其可变字段:

match expr {
  AtomExpr(atom_expr, ..) as node => {
    let ty = atom_expr.check_type(env)
    node.ty = Some(ty)  // 修改可变字段
    ty
  }
  // ...
}

这种设计避免了重新构建整个AST的开销,同时保持了函数式编程的风格。


完整编译流程展示​

经过词法分析、语法分析和类型检查三个阶段,我们的编译器前端已经能够将源代码转换为完全类型化的抽象语法树。让我们通过一个简单的例子来展示完整的过程:

源代码示例​

fn add(x: Int, y: Int) -> Int {
  return x + y;
}

编译输出:类型化AST​

利用 derive(ToJson)​ 功能,我们可以将最终的AST输出为JSON格式进行查看:

{
  "functions": {
    "add": {
      "name": "add",
      "params": [
        ["x", { "$tag": "Int" }],
        ["y", { "$tag": "Int" }]
      ],
      "ret_ty": { "$tag": "Int" },
      "body": [
        {
          "$tag": "Return",
          "0": {
            "$tag": "Binary",
            "0": "+",
            "1": {
              "$tag": "AtomExpr",
              "0": {
                "$tag": "Var",
                "0": "x",
                "ty": { "$tag": "Int" }
              },
              "ty": { "$tag": "Int" }
            },
            "2": {
              "$tag": "AtomExpr",
              "0": {
                "$tag": "Var",
                "0": "y",
                "ty": { "$tag": "Int" }
              },
              "ty": { "$tag": "Int" }
            },
            "ty": { "$tag": "Int" }
          }
        }
      ]
    }
  }
}

从这个JSON输出中,我们可以清楚地看到:

  1. 完整的函数签名:包括参数列表和返回类型
  2. 类型标记的AST节点:每个表达式都携带了类型信息
  3. 结构化的程序表示:为后续的代码生成阶段提供了清晰的数据结构

结语​

通过本篇文章,我们深入探讨了编译器前端的完整实现流程。从字符流到类型化的抽象语法树,我们见证了Moonbit语言在编译器构建中的独特优势:

核心收获​

  1. 模式匹配的威力:Moonbit的字符串模式匹配和结构化模式匹配极大简化了词法分析和语法分析的实现
  2. 函数式编程范式:loop​构造、环境链和不可变数据结构的结合,提供了既优雅又高效的解决方案
  3. 类型系统的表达力:通过枚举的可变字段和trait对象,我们能够构建既类型安全又灵活的数据结构
  4. 工程化特性:derive​功能、结构化错误处理和JSON序列化等特性,大大提升了开发效率

展望下篇​

在掌握了语法前端的实现之后,下篇文章将引导我们进入更加激动人心的代码生成阶段。我们将:

  • 深入了解LLVM中间表示的设计哲学
  • 探索Moonbit官方llvm.mbt​绑定库的使用方法
  • 实现从AST到LLVM IR的完整转换
  • 生成可执行的RISC-V汇编代码

编译器的构建是一个复杂而富有挑战性的过程,但正如我们在本篇中所展示的,Moonbit为这个过程提供了强大而优雅的工具。让我们在下篇中继续这段令人兴奋的编译器构建之旅。

资源推荐


函数式里的依赖注入:Reader Monad

· 阅读需 9 分钟

经常搞六边形架构的人也知道,为了保持核心业务逻辑的纯粹和独立,我们会把像数据库、外部 API 调用这些“副作用”放在“端口”和“适配器”里,然后通过 DI 的方式注入到应用层。可以说,经典的面向对象和分层架构,离不开 DI。

然后,当我想在 MoonBit 里做点事情的时候,我发现我不能呼吸了。

我们也想讲究一个入乡随俗,但是在 moonbit 这种函数味儿很浓郁的场地,没有类,没有接口,更没有我们熟悉的那一套 DI 容器。那我怎么做 DI?

我当时就在想,软件工程发展到至今已经约 57 年,真的没有在函数式编程里解决 DI 的方法吗?

有的兄弟,有的。只是它在函数式编程里也属于一种 monad:Reader Monad

什么是 Monad​

普通的函数就像一个流水线,你丢进去一袋面粉,然后直接跑到生产线末端,等着方便面出来。但这条流水线需要自动处理中间的所有复杂情况:

  • 没放面粉/“没有下单,期待发货”(null)
  • 面团含水量不够把压面机干卡了(抛出异常)
  • 配料机需要读取今天的生产配方,比如是红烧牛肉味还是香菇炖鸡味(读取外部配置)
  • 流水线末端的打包机需要记录今天打包了多少包(更新计数器)

Monad 就是专门管理这条复杂流水线的“总控制系统”。它把你的数据和处理流程的上下文一起打包,确保整个流程能顺畅、安全地进行下去。

在软件开发中,Monad 这一家子有几个常见的成员:

  • Option:处理“可能没有”的情况。盒子里要么有东西,要么是空的
  • Result:处理“可能会失败”的情况。盒子要么是绿的(成功),里面装着结果;要么是红的(失败),里面装着错误信息
  • State Monad:处理“需要修改状态”的情况。这个盒子在产出结果的同时,还会更新盒子侧面的一个计数器。或者说就是 React 里的 useState
  • Future(Promise):处理“未来才有”的情况。这个盒子给你一张“提货单”,承诺未来会把货给你
  • Reader Monad: 盒子可以随时查阅“环境”,但不能修改它

Reader Monad​

Reader Monad 的思想,最早可以追溯到上世纪90年代,在 Haskell 这种纯函数式编程语言的圈子里流行起来。当时大家为了坚守“函数纯度”这个铁律(即函数不能有副作用),就必须找到一种优雅的方式来让多个函数共享同一个配置环境,Reader Monad 就是为了解决这个矛盾而诞生的。

如今,它的应用场景已经非常广泛:

  • 应用配置管理:用来传递数据库连接池、API密钥、功能开关等全局配置
  • 请求上下文注入:在 Web 服务中,把当前登录的用户信息等打包成一个环境,供请求处理链上的所有函数使用
  • 实现六边形架构:在六边形(或端口与适配器)架构中,它被用来在核心业务逻辑(Domain/Application Layer)和外部基础设施(Infrastructure Layer)之间建立一道防火墙

简单来说,Reader Monad 就是一个专门处理只读环境依赖的工具。它要解决的就是这些问题:

  • 参数钻孔 (Parameter Drilling):我们不想把一个 Properties 层层传递
  • 逻辑与配置解耦:业务代码只关心“做什么”,而不用关心“配置从哪来”。这使得代码非常干净,且极易测试

核心方法​

一个 Reader 库通常包含以下几个核心部分。

Reader::pure​

就像是把一颗糖直接放进一个标准的午餐盒里。它把一个普通的值,包装成一个最简单的、不依赖任何东西的 Reader 计算。

pure 通常是流水线的打包机,它把你计算出的最终结果(一个普通值)重新放回 Reader “流水线”上,所谓“移除副作用”。

typealias @reader.Reader

// `pure` 创建一个不依赖环境的计算
let pure_reader : Reader[String, Int] = Reader::pure(100)

test {
  // 无论环境是什么 (比如 "hello"),结果都是 100
  assert_eq(pure_reader.run("hello"), 100)
}

Reader::bind​

这是流水线的“连接器”。例如把“和面”这一步和“压面”这一步连接起来,并确保它们能连成一条“生产线”。

为什么需要它? 为了自动化! 。bind 让这个过程全自动,你只管定义好每个步骤,它负责传递。

fnalias @reader.ask

// 步骤1: 定义一个 Reader,它的工作是从环境(一个Int)中读取值
let step1 : Reader[Int, Int] = ask()

// 步骤2: 定义一个函数,它接收一个数字,然后返回一个新的 Reader 计算
fn step2_func(n : Int) -> Reader[Int, Int] {
  Reader::pure(n * 2)
}

// 使用 bind 将两个步骤连接起来
let computation : Reader[Int, Int] = step1.bind(step2_func)

test {
  // 运行整个计算,环境是 5
  // 流程: step1 从环境得到 5 -> bind 把 5 交给 step2_func -> step2_func 计算 5*2=10 -> pure(10)
  assert_eq(computation.run(5), 10)
}

Reader::map​

就像是给午餐盒里的三明治换个标签。它只改变盒子里的东西(比如把薄荷塘换成酒心巧克力),但不动午餐盒本身。

很多时候我们只是想对结果做个简单转换,用 map 比用 bind 更直接,意图更清晰。

// `map` 只转换结果,不改变依赖
let reader_int : Reader[Unit, Int] = Reader::pure(5)

let reader_string : Reader[Unit, String] = reader_int.map(n => "Value is \{n}")

test {
  assert_eq(reader_string.run(()), "Value is 5")
}

ask​

ask 就像是流水线上的一个工人,随时可以抬头看一眼挂在墙上的“生产配方”。这是我们真正读取环境的唯一手段。

bind 只负责在幕后传递,但当你想知道“配方”里到底写了什么时,就必须用 ask 把它“问”出来。

// `ask` 直接获取环境
let ask_reader : Reader[String, String] = ask()

let result : String = ask_reader.run("This is the environment")

test {
  assert_eq(result, "This is the environment")
}

而我们接下来会经常用到的 asks,只是对 ask().map() 的封装。

DI 对比 Reader Monad​

搞个经典例子:开发一个 UserService,它需要一个 Logger 来记录日志,还需要一个 Database 来获取数据。

普通的 DI 我这里用我第二喜欢的 TypeScript 举例:

interface Logger {
  info(message: string): void
}
interface Database {
  getUserById(id: number): { name: string } | undefined
}

// 业务类通过构造函数声明其依赖
class UserService {
  constructor(
    private logger: Logger,
    private db: Database
  ) {}

  getUserName(id: number): string | undefined {
    this.logger.info(`Querying user with id: ${id}`)
    const user = this.db.getUserById(id)
    return user?.name
  }
}

// 创建依赖实例并注入
const myLogger: Logger = { info: (msg) => console.log(`[LOG] ${msg}`) }
const myDb: Database = {
  getUserById: (id) => (id === 1 ? { name: 'MoonbitLang' } : undefined)
}

const userService = new UserService(myLogger, myDb)
const userName = userService.getUserName(1) // "MoonbitLang"

// 一般来说我们会用一些库管理注入,不会手动实例化。例如 InversifyJS 亦或者是……Angular

而 Reader Monad 呢

fnalias @reader.asks

struct User {
  name : String
}

trait Logger {
  info(Self, String) -> Unit
}

trait Database {
  getUserById(Self, Int) -> User?
}

struct AppConfig {
  logger : &Logger
  db : &Database
}

fn getUserName(id : Int) -> Reader[AppConfig, String?] {
  asks(config => {
    config.logger.info("Querying user with id: \{id}")
    let user = config.db.getUserById(id)
    user.map(obj => obj.name)
  })
}

struct LocalDB {}

impl Database for LocalDB with getUserById(_, id) {
  if id == 1 {
    Some({ name: "MoonbitLang" })
  } else {
    None
  }
}

struct LocalLogger {}

impl Logger for LocalLogger with info(_, content) {
  println("\{content}")
}

test "Test UserName" {
  let appConfig = AppConfig::{ db: LocalDB::{  }, logger: LocalLogger::{  } }
  assert_eq(getUserName(1).run(appConfig).unwrap(), "MoonbitLang")
}

可以发现,getUserName 函数同样不持有任何依赖,它只是一个“计算描述”。

这个特性让 Reader Monad 成为了实现六边形架构的天作之合。在六边形架构里,核心原则是 “依赖倒置” ——核心业务逻辑不应该依赖具体的基础设施。

getUserName 的例子就是最好的体现。AppConfig 就是一个 Ports 集合

而 getUserName 这个核心业务逻辑,它只依赖 AppConfig 这个抽象,完全不知道背后到底是 MySQL 还是 PostgreSQL,还是一个假实现:一个 Mock DB

但它不能解决什么问题?状态修改。

Reader Monad 的环境永远是“只读”的。一旦注入,它在整个计算过程中都不能被改变。

如果你需要一个可变的状态,找它的兄弟 State Monad 吧。

也就是说,它的好处很明显:它可以在任意地方读取配置;

当然它的坏处也很明显:它只会读取。

简单的 i18n 工具库​

经常搞前端的人都知道,我们如果要搞 i18n,大概率会用上 i18next 这类库。它的核心玩法,通常是把一个 i18n 实例通过 React Context 注入到整个应用里,任何组件想用翻译,直接从 Context 里拿就行。所以这其实也可以是一种依赖注入。

回归初心了属于是,本来寻找 DI(Context) 的目的就是为了给 cli 工具支持 i18n。当然这里只是一个简单的演示。

首先,先安装依赖

moon add colmugx/reader

接着,我们来定义 i18n 库需要的环境和字典类型。

typealias String as Locale

typealias String as TranslationKey

typealias String as TranslationValue

typealias Map[TranslationKey, TranslationValue] as Translations

typealias Map[Locale, Translations] as Dict

struct I18nConfig {
  // 这里只是方便演示添加了 mut
  mut lang : Locale
  dict : Dict
}

接下来是翻译函数 t

fn t(key : TranslationKey) -> Reader[I18nConfig, TranslationValue] {
  asks(config => config.dict
    .get(config.lang)
    .map(lang_map => lang_map.get(key).unwrap_or(key))
    .unwrap_or(key))
}

完事了,看起来很简单是不是

接下来,假设我们的 CLI 工具需要根据操作系统的 LANG 环境变量来显示不同语言的欢迎信息。

fn welcome_message(content : String) -> Reader[I18nConfig, String] {
  t("welcome").bind(welcome_text => Reader::pure("\{welcome_text} \{content}"))
}

test {
  let dict : Dict = {
    "en_US": { "welcome": "Welcome To" },
    "zh_CN": { "welcome": "欢迎来到" },
  }

  // 假设你的系统语言 LANG 是 zh_CN
  let app_config = I18nConfig::{ lang: "zh_CN", dict }
  let msg = welcome_message("MoonbitLang")
  assert_eq(msg.run(app_config), "欢迎来到 MoonbitLang")

  // 切换语言
  app_config.lang = "en_US"
  assert_eq(msg.run(app_config), "Welcome To MoonbitLang")
}

欢迎来到 MoonbitLang

MoonBit pearls vol.4 用 MoonBit 探索协同式编程(下篇)

· 阅读需 12 分钟

本文旨在使用 MoonBit 语言的协同式编程库 moonchor,用多个例子阐释协同式编程的核心思想和基本用法。上篇文章中我们提到了如何通过一个书店应用展示moonbit在协同式编程里的实践。

案例研究:多副本 KVStore​

在本节中,我们将探讨一个更复杂的案例,使用 moonchor 实现多副本的 KVStore。我们依然只使用 moonchor 的核心 API,但会充分利用 MoonBit 的泛型和一等公民函数这两个特性。我们的目的是探索 MoonBit 的强大表达能力可以为协同式编程的带来多大的可能性。

基本实现​

首先做一些准备工作,定义客户端 Client 和服务器 Server 两个角色:

struct Server {} derive(Eq, Hash, Show)

struct Client {} derive(Eq, Hash, Show)

impl @moonchor.Location for Server with name(_) {
  "server"
}

impl @moonchor.Location for Client with name(_) {
  "client"
}

let server : Server = Server::{  }

let client : Client = Client::{  }

要实现一个 KVStore,例如 Redis,我们需要实现最基本的两个接口:get 和 put(对应 Redis 的 get 和 set)。最简单的实现就是用一个 Map 数据结构来存储键值对:

struct ServerState {
  db : Map[String, Int]
}

fn ServerState::new() -> ServerState {
  { db: {} }
}

对于 KVStore 而言,get 和 put 请求是客户端通过网络发送过来的,在接收到请求前,我们并不知道具体的请求是什么。所以我们需要定义一个请求类型 Request,它包含了请求的类型和参数:

enum Request {
  Get(String)
  Put(String, Int)
} derive(ToJson, FromJson)

为了方便,我们的 KVStore 只支持 String 类型的键和 Int 类型的值。接下来,我们定义一个 Response 类型,用于表示服务器对请求的响应:

typealias Int? as Response

响应是一个可选的整数。当请求是 Put 时,响应是 None;当请求是 Get 时,响应是键对应的值包裹上一个 Some,如果键不存在,则响应为 None。

fn handle_request(state : ServerState, request : Request) -> Response {
  match request {
    Request::Get(key) => state.db.get(key)
    Request::Put(key, value) => {
      state.db[key] = value
      None
    }
  }
}

我们的目标是定义两个函数 put 和 get 模拟客户端发起请求的过程。它们要做的事情分别是:

  1. 在 Client 处生成请求,包装键值对;
  2. 将请求发送给 Server;
  3. Server 使用 handle_request 函数处理请求;
  4. 将响应发送回 Client。

可以看到,put 和 get 函数的逻辑是相似的,我们可以把 2、3、4 三个过程抽象成一个函数,叫作 access_server。

async fn put_v1(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  key : String,
  value : Int
) -> Unit {
  let request = ctx.locally(client, _unwrapper => Request::Put(key, value))
  access_server_v1(ctx, request, state_at_server) |> ignore
}

async fn get_v1(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  key : String
) -> @moonchor.Located[Response, Client] {
  let request = ctx.locally(client, _unwrapper => Request::Get(key))
  access_server_v1(ctx, request, state_at_server)
}

async fn access_server_v1(
  ctx : @moonchor.ChoreoContext,
  request : @moonchor.Located[Request, Client],
  state_at_server : @moonchor.Located[ServerState, Server]
) -> @moonchor.Located[Response, Client] {
  let request_at_server = ctx.comm(client, server, request)
  let response = ctx.locally(server, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_server)
    let state = unwrapper.unwrap(state_at_server)
    handle_request(state, request)
  })
  ctx.comm(server, client, response)
}

这样我们的 KVStore 就完成了。我们可以写一个简单的 choreography 来测试它:

async fn kvstore_v1(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  put_v1(ctx, state_at_server, "key1", 42)
  put_v1(ctx, state_at_server, "key2", 41)
  let v1_at_client = get_v1(ctx, state_at_server, "key1")
  let v2_at_client = get_v1(ctx, state_at_server, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore v1" {
  let backend = @moonchor.make_local_backend([server, client])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v1, server))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v1, client))
}

这个程序的含义是,分别在 "key1" 和 "key2" 存储两个数字 42 和 41,然后从服务器获取这两个值并检查它们的和是否等于 83。如果有任何一个请求返回 None 或者计算结果不是 83,程序就会 panic。

双副本​

现在,考虑为 KVStore 增加容错功能。最简单的容错就是构建一个从副本,它与主副本存有相同的数据,并在处理 Get 请求时检查主从数据的一致性。

我们为从副本构建一个新的角色:

struct Backup {} derive(Eq, Hash, Show)

impl @moonchor.Location for Backup with name(_) {
  "backup"
}

let backup : Backup = Backup::{  }

定义一个函数用于检查一致性:这个函数会检查所有副本的响应是否一致,如果不一致,则 panic。

fn check_consistency(responses : Array[Response]) -> Unit {
  match responses.pop() {
    None => return
    Some(f) =>
      for res in responses {
        if res != f {
          panic()
        }
      }
  }
}

其余的大部分内容都不需要修改,只要在 access_server 函数中增加对副本的处理即可。新的 access_server_v2 的逻辑是,Server 接收到请求后,将请求转发给 Backup;然后 Server 和 Backup 分别处理请求;Backup 处理完请求后发回给 Server,Server 对两个结果进行一致性检验。

async fn put_v2(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup],
  key : String,
  value : Int
) -> Unit {
  let request = ctx.locally(client, _unwrapper => Request::Put(key, value))
  access_server_v2(ctx, request, state_at_server, state_at_backup) |> ignore
}

async fn get_v2(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup],
  key : String
) -> @moonchor.Located[Response, Client] {
  let request = ctx.locally(client, _unwrapper => Request::Get(key))
  access_server_v2(ctx, request, state_at_server, state_at_backup)
}

async fn access_server_v2(
  ctx : @moonchor.ChoreoContext,
  request : @moonchor.Located[Request, Client],
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup]
) -> @moonchor.Located[Response, Client] {
  let request_at_server = ctx.comm(client, server, request)
  let request_at_backup = ctx.comm(server, backup, request_at_server)
  let response_at_backup = ctx.locally(backup, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_backup)
    let state = unwrapper.unwrap(state_at_backup)
    handle_request(state, request)
  })
  let backup_response_at_server = ctx.comm(backup, server, response_at_backup)
  let response_at_server = ctx.locally(server, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_server)
    let state = unwrapper.unwrap(state_at_server)
    let response = handle_request(state, request)
    let backup_response = unwrapper.unwrap(backup_response_at_server)
    check_consistency([response, backup_response])
    response
  })
  ctx.comm(server, client, response_at_server)
}

和刚才一样,我们可以写一个简单的 choreography 来测试它:

async fn kvstore_v2(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  let state_at_backup = ctx.locally(backup, _unwrapper => ServerState::new())
  put_v2(ctx, state_at_server, state_at_backup, "key1", 42)
  put_v2(ctx, state_at_server, state_at_backup, "key2", 41)
  let v1_at_client = get_v2(ctx, state_at_server, state_at_backup, "key1")
  let v2_at_client = get_v2(ctx, state_at_server, state_at_backup, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore 2.0" {
  let backend = @moonchor.make_local_backend([server, client, backup])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, server) )
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, client) )
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, backup) )
}

利用高阶函数抽象复制策略​

在双副本实现过程中,出现了一些耦合的代码:Server 处理请求、备份请求、检查结果一致性的代码放在了一起。

利用 MoonBit 的高阶函数特性,我们可以把复制策略从具体处理过程中抽象出来。我们分析一下什么是复制策略。复制策略应该包含一个过程,即服务器拿到请求后如何利用各个副本处理它的方式。关键在于,复制策略本身是和请求无关的,应该被从具体请求处理过程中剥离出来。这样的话,我们就能让复制策略成为可替换的部分,便于日后能轻易地在不同的复制策略之间进行切换,或者实现新的复制策略。

当然,真实世界的复制策略是非常复杂的,往往很难清晰地从处理流程中剥离出来。在这个例子中,我们为了简化问题,专注于 moonchor 的编程能力,直接将复制策略定义为 Server 在接收到请求后决定如何处理请求的函数。我们可以用一个类型别名来定义它:

typealias async (@moonchor.ChoreoContext, @moonchor.Located[Request, Server]) -> @moonchor.Located[
  Response,
  Server,
] as ReplicationStrategy

接下来,我们就可以简化 access_server 的实现了。我们将策略作为参数传递进去:

async fn access_server_v3(
  ctx : @moonchor.ChoreoContext,
  request : @moonchor.Located[Request, Client],
  strategy : ReplicationStrategy
) -> @moonchor.Located[Response, Client] {
  let request_at_server = ctx.comm(client, server, request)
  let response = strategy(ctx, request_at_server)
  ctx.comm(server, client, response)
}

async fn put_v3(
  ctx : @moonchor.ChoreoContext,
  strategy : ReplicationStrategy,
  key : String,
  value : Int
) -> Unit {
  let request = ctx.locally(client, _unwrapper => Request::Put(key, value))
  access_server_v3(ctx, request, strategy) |> ignore
}

async fn get_v3(
  ctx : @moonchor.ChoreoContext,
  strategy : ReplicationStrategy,
  key : String
) -> @moonchor.Located[Response, Client] {
  let request = ctx.locally(client, _unwrapper => Request::Get(key))
  access_server_v3(ctx, request, strategy)
}

这样一来,复制策略被成功从处理请求的逻辑中抽象出来了。下面,我们重新实现一遍双副本的复制策略:

async fn double_replication_strategy(
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup],
) -> ReplicationStrategy {
  fn(
    ctx : @moonchor.ChoreoContext,
    request_at_server : @moonchor.Located[Request, Server]
  ) {
    let request_at_backup = ctx.comm(server, backup, request_at_server)
    let response_at_backup = ctx.locally(backup, fn(unwrapper) {
      let request = unwrapper.unwrap(request_at_backup)
      let state = unwrapper.unwrap(state_at_backup)
      handle_request(state, request)
    })
    let backup_response = ctx.comm(backup, server, response_at_backup)
    ctx.locally(server, fn(unwrapper) {
      let request = unwrapper.unwrap(request_at_server)
      let state = unwrapper.unwrap(state_at_server)
      let res = handle_request(state, request)
      check_consistency([unwrapper.unwrap(backup_response), res])
      res
    })
  }
}

注意看 double_replication_strategy 的函数签名,它返回一个 ReplicationStrategy 类型的函数。只要提供两个参数,double_replication_strategy 就能构造出一个新的复制策略。至此,我们成功利用高阶函数抽象出了复制策略,这个特性在协同式编程中叫作高阶 choreography。

同样的,我们可以写一个简单的 choreography 来测试它:

async fn kvstore_v3(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  let state_at_backup = ctx.locally(backup, _unwrapper => ServerState::new())
  let strategy = double_replication_strategy(state_at_server, state_at_backup)
  put_v3(ctx, strategy, "key1", 42)
  put_v3(ctx, strategy, "key2", 41)
  let v1_at_client = get_v3(ctx, strategy, "key1")
  let v2_at_client = get_v3(ctx, strategy, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore 3.0" {
  let backend = @moonchor.make_local_backend([server, client, backup])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, server))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, client))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, backup))
}

利用参数化多态实现角色多态​

如果要进一步实现新的复制策略,例如三副本,我们需要定义两个新的 Backup 类型以做区分:

struct Backup1 {} derive(Eq, Hash, Show)

impl @moonchor.Location for Backup1 with name(_) {
  "backup1"
}

let backup1 : Backup1 = Backup1::{  }

struct Backup2 {} derive(Eq, Hash, Show)

impl @moonchor.Location for Backup2 with name(_) {
  "backup2"
}

let backup2 : Backup2 = Backup2::{  }

接下来需要修改 access_server 的核心逻辑。我们立刻发现了问题,为了让 Backup1 和 Backup2 都处理一遍请求并且得到响应,需要将以下几条语句重复:let request = unwrapper.unwrap(request_at_backup); let state = unwrapper.unwrap(state_at_backup); handle_request(state, request)。重复代码是坏味道,应当被抽象出来。此时,moonchor 的「角色作为类型」优势就体现出来了,我们可以利用 MoonBit 的参数化多态,将从副本处理逻辑抽象成一个多态函数 do_backup,它接收一个角色类型参数 B,表示从副本的角色:

async fn[B : @moonchor.Location] do_backup(
  ctx : @moonchor.ChoreoContext,
  request_at_server : @moonchor.Located[Request, Server],
  backup : B,
  state_at_backup : @moonchor.Located[ServerState, B]
) -> @moonchor.Located[Response, Server] {
  let request_at_backup = ctx.comm(server, backup, request_at_server)
  let response_at_backup = ctx.locally(backup, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_backup)
    let state = unwrapper.unwrap(state_at_backup)
    handle_request(state, request)
  })
  ctx.comm(backup, server, response_at_backup)
}

如此一来,我们就能随心所欲地实现双副本或者三副本的复制策略了。对于三副本策略,只需在 triple_replication_strategy 返回的函数内调用 do_backup 两次即可:

async fn triple_replication_strategy(
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup1 : @moonchor.Located[ServerState, Backup1],
  state_at_backup2 : @moonchor.Located[ServerState, Backup2]
) -> ReplicationStrategy {
  fn(
    ctx : @moonchor.ChoreoContext,
    request_at_server : @moonchor.Located[Request, Server]
  ) {
    let backup_response1 = do_backup(
      ctx, request_at_server, backup1, state_at_backup1,
    )
    let backup_response2 = do_backup(
      ctx, request_at_server, backup2, state_at_backup2,
    )
    ctx.locally(server, fn(unwrapper) {
      let request = unwrapper.unwrap(request_at_server)
      let state = unwrapper.unwrap(state_at_server)
      let res = handle_request(state, request)
      check_consistency([
        unwrapper.unwrap(backup_response1),
        unwrapper.unwrap(backup_response2),
        res,
      ])
      res
    })
  }
}

由于我们成功完成了复制策略和访问过程的分离,access_server、put、get 函数不需要任何修改。让我们对最终的 KVStore 进行测试:

async fn kvstore_v4(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  let state_at_backup1 = ctx.locally(backup1, _unwrapper => ServerState::new())
  let state_at_backup2 = ctx.locally(backup2, _unwrapper => ServerState::new())
  let strategy = triple_replication_strategy(
    state_at_server, state_at_backup1, state_at_backup2,
  )
  put_v3(ctx, strategy, "key1", 42)
  put_v3(ctx, strategy, "key2", 41)
  let v1_at_client = get_v3(ctx, strategy, "key1")
  let v2_at_client = get_v3(ctx, strategy, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore 4.0" {
  let backend = @moonchor.make_local_backend([server, client, backup1, backup2])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, server))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, client))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, backup1))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, backup2))
}

至此,我们完成了多副本 KVStore 的构建。在这个例子中,我们没有手动使用任何 send 和 recv 来表达分布式节点间的交互,而是通过 moonchor 的协同式编程能力实现了所有通信和同步过程,避免可能的类型错误、死锁和显式同步问题。

结语

在这篇文章中,我们借助 moonchor 体验了协同式编程的魅力,还见识了 MoonBit 强大的表达能力。关于协同式编程的更多细节,可以参考 Haskell 的库 HasChor、Choral 语言、moonchor 的源码。想要自己尝试使用 moonchor,可以通过 moon add Milky2018/moonchor@0.15.0 命令安装。

MoonBit Pearls Vol.04: 用MoonBit探索协同式编程

· 阅读需 26 分钟

传统的分布式程序设计是非常痛苦的,其中一个重要的因素是,很多整体的逻辑需要拆散到各个分布式节点中实现,分散的实现使得程序难以调试、难以理解,并且无法享用编程语言提供的类型检查能力。Choreographic Programming,即协同式编程,提供了一种整体的视角,允许开发者编写需要多个参与者协同工作的单一程序,然后将这个整体程序分别投射到各个参与者,最终实现协同工作的效果。

协同式编程通过两种不同的方式实现:其一是作为一种全新的编程语言,例如 Choral,开发者编写 Choral 程序,然后用编译器将这个单体程序编译到各个参与者专属的 Java 程序;其二是作为一个库,例如 HasChor,直接利用 Haskell 的类型系统就能实现协同式编程的静态性质,并且完美兼容 Haskell 的生态。MoonBit 的函数式编程特性和强大的类型系统使得它很适合用于构建协同式编程的库。

本文旨在使用 MoonBit 语言的协同式编程库 moonchor,用多个例子阐释协同式编程的核心思想和基本用法。

导览:书店应用​

让我们考察一个书店应用,该应用包含两个角色:买家和卖家,其核心逻辑如下:

  1. 买家向卖家发送想要购买的书的标题;
  2. 卖家通过查询数据库告诉买家书的价格;
  3. 买家决定是否购买书籍;
  4. 如果买家决定购买,卖家从数据库中扣除书籍的库存并发送预期送达日期给买家;
  5. 否则,交互中止。

传统实现​

我们在此不关心实现细节,只关心核心逻辑,使用 send 和 recv 函数来表示发送和接收消息。按照传统的实现方式,我们需要为买家和卖家分别开发两个应用。在表示这些应用之前,我们假设已经存在一些函数和类型:

fn get_title() -> String {
  "Homotopy Type Theory"
}

fn get_price(title : String) -> Int {
  50
}

fn get_budget() -> Int {
  100
}

fn get_delivery_date(title : String) -> String {
  "2025-10-01"
}

enum Role {
  Buyer
  Seller
}

async fn[T] send(msg : T, target : Role) -> Unit {
  ...
}

async fn[T] recv(source : Role) -> T {
  ...
}

买家的应用如下:

async fn book_buyer() -> Unit {
  let title = get_title()
  send(title, Seller)
  let price = recv(Seller)
  if price <= get_budget() {
    send(true, Seller)
    let delivery_date = recv(Seller)
    println("The book will be delivered on: \{delivery_date}")
  } else {
    send(false, Seller)
  }
}

卖家的应用如下:

async fn book_seller() -> Unit {
  let title = recv(Buyer)
  let price = get_price(title)
  send(price, Buyer)
  let decision = recv(Buyer)
  if decision {
    let delivery_date = get_delivery_date(title)
    send(delivery_date, Buyer)
  }
}

这两个应用至少有以下几个问题:

  1. 无法保证类型安全:注意到 send 和 recv 都是泛型函数,只有当发送和接收的类型一致时,才能保证类型安全;否则,可能会在序列化、反序列化过程发生运行时错误。而编译期无法检查这种类型安全性,因为编译器无法知道每个 send 对应哪个 recv,只能寄希望于开发者不会写错。
  2. 可能导致死锁:万一买家程序的某个 send 语句漏写了,买家和卖家可能会同时等待对方的消息;或者在网络交互时,某个买家连接暂时断开了,卖家也会一直等待买家的消息。上述两种情况都导致死锁。
  3. 需要显式同步:买家为了向卖家传达是否要购买的决定,必须显式地发送一个 Bool 类型的消息。后续的协同过程需要保证买家和卖家在 if price <= get_budget() 和 if decision 这两个位置走进相同的分支,而这一特点也是无法在编译期保证的。

导致这些问题的根本原因是我们将一个整体的协同逻辑按照实现的需求拆成了两个独立的部分。接下来,我们看看使用协同式编程如何解决上述问题。

moonchor 实现​

使用协同式编程,我们可以将买家和卖家的逻辑写在同一个函数中,然后让它根据调用该函数时不同的参数表现出不同的行为。我们使用 moonchor 中的 API 来定义买家和卖家的角色。在 moonchor 中,角色被定义为 trait Location。为了提供更好的静态性质,角色不仅是值,同时还是一个独特的类型,该类型需要实现 Location 这个 trait。

struct Buyer {} derive(Eq, Show, Hash)

impl @moonchor.Location for Buyer with name(_) {
  "buyer"
}

struct Seller {} derive(Eq, Show, Hash)

impl @moonchor.Location for Seller with name(_) {
  "seller"
}

let buyer : Buyer = Buyer::{  }

let seller : Seller = Seller::{  }

可以看见,我们定义的 Buyer 和 Seller 类型不包含任何字段。实现 Location trait 的类型只需要提供一个 name 方法,返回一个字符串作为角色的名称。这个 name 方法非常重要,它标识着角色的身份属性,并在类型检查无法保证类型安全时,提供最终检查手段。不要为不同的角色设置相同的名称,否则会导致意外的运行时错误。我们将在后文了解到类型如何保证一定程度的安全性,以及为什么仅依靠类型是不够的。

接下来,我们定义书店应用的核心逻辑,它被称作一个 choreography:

async fn bookshop(ctx : @moonchor.ChoreoContext) -> Unit {
  let title_at_buyer = ctx.locally(buyer, _unwrapper => get_title())
  let title_at_seller = ctx.comm(buyer, seller, title_at_buyer)
  let price_at_seller = ctx.locally(seller, fn(unwrapper) {
    let title = unwrapper.unwrap(title_at_seller)
    get_price(title)
  })
  let price_at_buyer = ctx.comm(seller, buyer, price_at_seller)
  let decision_at_buyer = ctx.locally(buyer, fn(unwrapper) {
    let price = unwrapper.unwrap(price_at_buyer)
    price < get_budget()
  })
  if ctx.broadcast(buyer, decision_at_buyer) {
    let delivery_date_at_seller = ctx.locally(seller, unwrapper => get_delivery_date(
      unwrapper.unwrap(title_at_seller),
    ))
    let delivery_date_at_buyer = ctx.comm(
      seller, buyer, delivery_date_at_seller,
    )
    ctx.locally(buyer, fn(unwrapper) {
      let delivery_date = unwrapper.unwrap(delivery_date_at_buyer)
      println("The book will be delivered on \{delivery_date}")
    })
    |> ignore
  }
}

这个程序稍微有点长,我们先逐行分析一下。

函数的参数 ctx: @moonchor.ChoreoContext 是 moonchor 给应用提供的上下文对象,它包含了协同式编程在应用侧的所有接口。首先,我们使用 ctx.locally 执行一个仅在买家角色处需要执行的操作 get_title()。ctx.locally 的第一个参数是角色,第二个参数是一个闭包,闭包的内容就是需要执行的参数,返回值被包装后作为 ctx.locally 的返回值。在这里,get_title() 的返回值是 String 类型,而 title_at_buyer 的类型是 @moonchor.Located[String, Buyer],表示这个值位于买家这个角色,无法被其它角色使用。当你试图在卖家角色中使用 title_at_buyer 时,编译器会报错,告诉你 Buyer 和 Seller 不是同一个类型。

接下来,买家需要将书名发送给卖家,我们使用 ctx.comm 来实现这个操作。ctx.comm 的第一个参数是发送者角色,第二个参数是接收者角色,第三个参数是发送的内容。在这里,ctx.comm 的返回值 title_at_seller 的类型是 @moonchor.Located[String, Seller],表示这个值位于卖家角色。你已经猜到了,ctx.comm 对应的操作正是 send 和 recv。但这里,类型得到了保障:ctx.comm 是一个泛型函数,它保证1)发送和接受的消息是同一个类型;2)发送者和接收者的角色对应为参数类型和返回值类型的类型参数,即 @moonchor.Located[T, Sender] 和 @moonchor.Located[T, Receiver]。

再往下,卖家开始通过查询数据库获取书的价格。在这一步我们用到了 ctx.locally 传递给闭包的参数 unwrapper。这个参数是一个用于为 Located 类型解包的对象,它的类型签名中也包含一个角色类型参数,我们通过 Unwrapper::unwrap 方法的签名即可看懂它是如何工作的:fn[T, L] Unwrapper::unwrap(_ : Unwrapper[L], v : Located[T, L]) -> T。也就是说,ctx.locally(buyer, unwrapper => ...) 中的 unwrapper 的类型是 Unwrapper[Buyer],而 title_at_seller 的类型是 Located[String, Seller],因此 unwrapper.unwrap(title_at_seller) 的结果类型是 String。这就是我们可以在闭包中使用 title_at_seller 而不能使用 title_at_buyer 的原因。

Knowledge of Choice​

在后续的流程中,如何解决显式同步问题是一个关键点,以至于我们要单独用一个小节来说明。在协同式编程中,这个问题被称作 Knowledge of Choice(选择知识)。在上面的例子中,买家需要知道是否购买书籍,而卖家需要知道买家是否购买书籍。我们使用 ctx.broadcast 来实现这个功能。

ctx.broadcast 的第一个参数是发送者的角色,第二个参数是需要共享给所有其它角色的消息。在这个例子中,买家和卖家都需要知道买家是否购买书籍,因此买家要将这一决定 decision_at_buyer 通过 ctx.broadcast 发送给所有参与者(在这里只有卖家)。有趣的是,这个 broadcast 的返回值是一个普通类型而非 Located 类型,这意味着它可以被所有角色使用,并且直接在顶层使用而不需要在 locally 中用 unwrapper 解包。因此,我们能够利用 MoonBit 本身的 if 条件语句来编写后续流程,从而保证买家和卖家在 if 分支中走入相同的分支。

从名字可以看出,ctx.broadcast 的作用是在整个 choreography 中广播一个值。它不仅可以广播一个 Bool 类型,也可以广播任意其它类型。它的结果不仅可以应用于 if 条件语句,也可以用于 while 循环或者任何其它需要公共知识的地方。

启动代码​

这样一个 choreography 怎样运行呢?moonchor 提供了 run_choreo 函数来启动一个 choreography。目前,由于 MoonBit 的多后端特性,提供稳定的、可移植的 TCP 服务器和跨进程通信接口是一项挑战,因此我们将使用协程和通道来探寻 choreography 的真正运行过程。完整的启动代码如下:

test "Blog: bookshop" {
  let backend = @moonchor.make_local_backend([buyer, seller])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, bookshop, buyer) )
  @toolkit.run_async(() => @moonchor.run_choreo(backend, bookshop, seller) )
}

上述代码启动了两个协程,分别在买家和卖家处执行同一个 choreography。也可以理解为,bookshop 这个函数被投射成(也被称为 EPP,端点投射)了「买家版」和「卖家版」两个完全不同的版本。在上面的例子中,run_choreo 的第一个参数是一个 Backend 类型的对象,它提供了协同式编程所需的底层通信机制。我们使用 make_local_backend 函数创建了一个本地后端(不要和刚刚提到的 MoonBit 多后端混淆),这个后端可以在本地进程中运行,使用 peter-jerry-ye/async/channel 提供的通道 API 作为通信基础。在未来,moonchor 还会提供更多的后端实现,例如 HTTP。

API 和部分原理​

我们已经对协同式编程和 moonchor 有了初步的了解。接下来,我们正式引入刚刚用到的 API 以及一些没有用到的 API,并且介绍它们的部分原理。

角色​

在 moonchor 中,我们通过实现 Location 这个 trait 来定义角色。该 trait 的声明如下:

pub(open) trait Location: Show + Hash {
  name(Self) -> String
}

Location 的 trait object 实现了 Eq:

impl Eq for &Location with op_equal(self, other) {
  self.name() == other.name()
}

如果两个角色的 name 方法返回相同的字符串,那么它们被认为是同一个角色,否则就不是。在判断某个值是否是某个角色时,name 方法是最终裁定者。也就是说,可以存在类型相同但实际上不是同一角色的值。这个特性在处理动态生成的角色时是尤其重要的。比如在书店例子中,买家有可能不止一个,卖家需要同时处理多个买家请求,并且根据服务器接收到的连接来动态生成买家角色。此时,买家的类型定义如下:

struct DynamicBuyer {
  id : String
} derive(Eq, Show, Hash)

impl @moonchor.Location for DynamicBuyer with name(self) {
  "buyer-\{self.id}"
}

Located Values​

因为 choreography 中会同时出现位于不同角色的值,因此我们需要某种手段来区分每个值都是位于哪个角色之处的。在 moonchor 中,这个用 Located[T, L] 这个类型表示位于角色 L 处的类型为 T 的值。

type Located[T, L]

type Unwrapper[L]

构建一个 Located Value 的方式是通过 ChoreoContext::locally 或 ChoreoContext::comm。这两个函数都会返回一个 Located 值。

使用一个 Located Value 的方式是通过 Unwrapper 对象的 unwrap 方法。这些内容在上面的书店应用中已经展示过了,不作赘述。

局部计算​

我们在例子中见到的最常见的 API 即为 ChoreoContext::locally,它用于在某个角色处执行一个局部计算动作。其签名如下:

type ChoreoContext

fn[T, L : Location] locally(
  self : ChoreoContext,
  location : L,
  computation : (Unwrapper[L]) -> T
) -> Located[T, L] {
  ...
}

该 API 表示会在 location 这个角色处执行 computation 这个闭包,并将计算结果包装成一个 Located Value。computation 闭包的唯一参数是一个解包器对象,类型为 Unwrapper[L],它在闭包中用于将 Located[T, L] 类型的值解包成 T 类型。这个 API 的作用是将计算的结果绑定到某个角色上,确保该值只能在该角色处使用。如果试图在其它角色处使用这个值,或用这个解包器处理其它角色的值,编译器会报错。

通信​

ChoreoContext::comm API 用于将一个值从一个角色发送到另一个角色。其签名如下:

trait Message: ToJson + @json.FromJson {}

async fn[T : Message, From : Location, To : Location] comm(
  self : ChoreoContext,
  from : From,
  to : To,
  value : Located[T, From]
) -> Located[T, To] {
  ...
}

发送和接收通常意味着需要序列化和反序列化过程。在 moonchor 目前的实现中,为了方便,使用 Json 作为消息的物理载体。未来可能会改用字节流作为更高效和通用的物理载体。

ChoreoContext::comm 有三个类型参数,除了要发送的消息类型,还有发送方和接收方的角色类型 From 和 To。这两个类型刚好对应了该方法的 from 参数、to 参数,以及 value 参数和返回值的类型。这保证了发送方和接收方在该消息序列化、反序列化的类型安全性,并且保证发送和接收行为必然会配对,不会因疏忽导致死锁。

广播​

当需要在多个角色之间共享一个值时,我们使用 ChoreoContext::broadcast API 让某个角色将一个值广播给所有其它角色。其签名如下:

async fn[T : Message, L : Location] ChoreoContext::broadcast(
  self : ChoreoContext,
  loc : L,
  value : Located[T, L]
) -> T {
  ...
}

广播和通信的 API 很相似,除了两点不同:

  1. 广播不需要指明接收方的角色,默认是该 choreography 中的所有角色;
  2. 广播的返回值并非 Located Value,而是消息本身的类型。

这两个特点揭示了广播的目的:所有角色都能访问到同一个值,从而在 choreography 的顶层对该值进行操作而不是局限在 ChoreoContext::locally 方法内部。例如在书店例子中,买家和卖家需要对「是否购买」这一决定达成共识,以确保后续的流程仍然保持一致。

后端和运行​

运行一个 choreography 的 API 如下:

type Backend

typealias async (ChoreoContext) -> T as Choreo[T]

async fn[T, L : Location] run_choreo(
  backend : Backend,
  choreography : Choreo[T],
  role : L
) -> T {
  ...
}

它接收三个参数:一个后端、一个用户编写的 choreography 和一个待运行的角色。后端包含了通信机制的具体实现,待运行的角色则是指定这个 choreography 要在哪个位置执行。比如之前的例子中,买家的程序需要在此处传递一个 Buyer 类型的值,而卖家需要传递 Seller 类型的值。

moonchor 提供了一个基于协程和通道的本地后端:

fn make_local_backend(locations : Array[&Location]) -> Backend {
  ...
}

这个函数为参数中的所有角色之间构建通信通道,提供具体的通信实现,即 send 和 recv 方法。尽管本地后端只能用于单体并发程序而非真正的分布式应用程序,但它的实现是可插拔的。只要拥有了基于稳定的网络通信 API 实现的其它后端,moonchor 就能轻松用于构建分布式程序了。

(可选阅读)案例研究:多副本 KVStore​

在本节中,我们将探讨一个更复杂的案例,使用 moonchor 实现多副本的 KVStore。我们依然只使用 moonchor 的核心 API,但会充分利用 MoonBit 的泛型和一等公民函数这两个特性。我们的目的是探索 MoonBit 的强大表达能力可以为协同式编程的带来多大的可能性。

基本实现​

首先做一些准备工作,定义客户端 Client 和服务器 Server 两个角色:

struct Server {} derive(Eq, Hash, Show)

struct Client {} derive(Eq, Hash, Show)

impl @moonchor.Location for Server with name(_) {
  "server"
}

impl @moonchor.Location for Client with name(_) {
  "client"
}

let server : Server = Server::{  }

let client : Client = Client::{  }

要实现一个 KVStore,例如 Redis,我们需要实现最基本的两个接口:get 和 put(对应 Redis 的 get 和 set)。最简单的实现就是用一个 Map 数据结构来存储键值对:

struct ServerState {
  db : Map[String, Int]
}

fn ServerState::new() -> ServerState {
  { db: {} }
}

对于 KVStore 而言,get 和 put 请求是客户端通过网络发送过来的,在接收到请求前,我们并不知道具体的请求是什么。所以我们需要定义一个请求类型 Request,它包含了请求的类型和参数:

enum Request {
  Get(String)
  Put(String, Int)
} derive(ToJson, FromJson)

为了方便,我们的 KVStore 只支持 String 类型的键和 Int 类型的值。接下来,我们定义一个 Response 类型,用于表示服务器对请求的响应:

typealias Int? as Response

响应是一个可选的整数。当请求是 Put 时,响应是 None;当请求是 Get 时,响应是键对应的值包裹上一个 Some,如果键不存在,则响应为 None。

fn handle_request(state : ServerState, request : Request) -> Response {
  match request {
    Request::Get(key) => state.db.get(key)
    Request::Put(key, value) => {
      state.db[key] = value
      None
    }
  }
}

我们的目标是定义两个函数 put 和 get 模拟客户端发起请求的过程。它们要做的事情分别是:

  1. 在 Client 处生成请求,包装键值对;
  2. 将请求发送给 Server;
  3. Server 使用 handle_request 函数处理请求;
  4. 将响应发送回 Client。

可以看到,put 和 get 函数的逻辑是相似的,我们可以把 2、3、4 三个过程抽象成一个函数,叫作 access_server。

async fn put_v1(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  key : String,
  value : Int
) -> Unit {
  let request = ctx.locally(client, _unwrapper => Request::Put(key, value))
  access_server_v1(ctx, request, state_at_server) |> ignore
}

async fn get_v1(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  key : String
) -> @moonchor.Located[Response, Client] {
  let request = ctx.locally(client, _unwrapper => Request::Get(key))
  access_server_v1(ctx, request, state_at_server)
}

async fn access_server_v1(
  ctx : @moonchor.ChoreoContext,
  request : @moonchor.Located[Request, Client],
  state_at_server : @moonchor.Located[ServerState, Server]
) -> @moonchor.Located[Response, Client] {
  let request_at_server = ctx.comm(client, server, request)
  let response = ctx.locally(server, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_server)
    let state = unwrapper.unwrap(state_at_server)
    handle_request(state, request)
  })
  ctx.comm(server, client, response)
}

这样我们的 KVStore 就完成了。我们可以写一个简单的 choreography 来测试它:

async fn kvstore_v1(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  put_v1(ctx, state_at_server, "key1", 42)
  put_v1(ctx, state_at_server, "key2", 41)
  let v1_at_client = get_v1(ctx, state_at_server, "key1")
  let v2_at_client = get_v1(ctx, state_at_server, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore v1" {
  let backend = @moonchor.make_local_backend([server, client])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v1, server))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v1, client))
}

这个程序的含义是,分别在 "key1" 和 "key2" 存储两个数字 42 和 41,然后从服务器获取这两个值并检查它们的和是否等于 83。如果有任何一个请求返回 None 或者计算结果不是 83,程序就会 panic。

双副本​

现在,考虑为 KVStore 增加容错功能。最简单的容错就是构建一个从副本,它与主副本存有相同的数据,并在处理 Get 请求时检查主从数据的一致性。

我们为从副本构建一个新的角色:

struct Backup {} derive(Eq, Hash, Show)

impl @moonchor.Location for Backup with name(_) {
  "backup"
}

let backup : Backup = Backup::{  }

定义一个函数用于检查一致性:这个函数会检查所有副本的响应是否一致,如果不一致,则 panic。

fn check_consistency(responses : Array[Response]) -> Unit {
  match responses.pop() {
    None => return
    Some(f) =>
      for res in responses {
        if res != f {
          panic()
        }
      }
  }
}

其余的大部分内容都不需要修改,只要在 access_server 函数中增加对副本的处理即可。新的 access_server_v2 的逻辑是,Server 接收到请求后,将请求转发给 Backup;然后 Server 和 Backup 分别处理请求;Backup 处理完请求后发回给 Server,Server 对两个结果进行一致性检验。

async fn put_v2(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup],
  key : String,
  value : Int
) -> Unit {
  let request = ctx.locally(client, _unwrapper => Request::Put(key, value))
  access_server_v2(ctx, request, state_at_server, state_at_backup) |> ignore
}

async fn get_v2(
  ctx : @moonchor.ChoreoContext,
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup],
  key : String
) -> @moonchor.Located[Response, Client] {
  let request = ctx.locally(client, _unwrapper => Request::Get(key))
  access_server_v2(ctx, request, state_at_server, state_at_backup)
}

async fn access_server_v2(
  ctx : @moonchor.ChoreoContext,
  request : @moonchor.Located[Request, Client],
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup]
) -> @moonchor.Located[Response, Client] {
  let request_at_server = ctx.comm(client, server, request)
  let request_at_backup = ctx.comm(server, backup, request_at_server)
  let response_at_backup = ctx.locally(backup, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_backup)
    let state = unwrapper.unwrap(state_at_backup)
    handle_request(state, request)
  })
  let backup_response_at_server = ctx.comm(backup, server, response_at_backup)
  let response_at_server = ctx.locally(server, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_server)
    let state = unwrapper.unwrap(state_at_server)
    let response = handle_request(state, request)
    let backup_response = unwrapper.unwrap(backup_response_at_server)
    check_consistency([response, backup_response])
    response
  })
  ctx.comm(server, client, response_at_server)
}

和刚才一样,我们可以写一个简单的 choreography 来测试它:

async fn kvstore_v2(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  let state_at_backup = ctx.locally(backup, _unwrapper => ServerState::new())
  put_v2(ctx, state_at_server, state_at_backup, "key1", 42)
  put_v2(ctx, state_at_server, state_at_backup, "key2", 41)
  let v1_at_client = get_v2(ctx, state_at_server, state_at_backup, "key1")
  let v2_at_client = get_v2(ctx, state_at_server, state_at_backup, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore 2.0" {
  let backend = @moonchor.make_local_backend([server, client, backup])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, server) )
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, client) )
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, backup) )
}

利用高阶函数抽象复制策略​

在双副本实现过程中,出现了一些耦合的代码:Server 处理请求、备份请求、检查结果一致性的代码放在了一起。

利用 MoonBit 的高阶函数特性,我们可以把复制策略从具体处理过程中抽象出来。我们分析一下什么是复制策略。复制策略应该包含一个过程,即服务器拿到请求后如何利用各个副本处理它的方式。关键在于,复制策略本身是和请求无关的,应该被从具体请求处理过程中剥离出来。这样的话,我们就能让复制策略成为可替换的部分,便于日后能轻易地在不同的复制策略之间进行切换,或者实现新的复制策略。

当然,真实世界的复制策略是非常复杂的,往往很难清晰地从处理流程中剥离出来。在这个例子中,我们为了简化问题,专注于 moonchor 的编程能力,直接将复制策略定义为 Server 在接收到请求后决定如何处理请求的函数。我们可以用一个类型别名来定义它:

typealias async (@moonchor.ChoreoContext, @moonchor.Located[Request, Server]) -> @moonchor.Located[
  Response,
  Server,
] as ReplicationStrategy

接下来,我们就可以简化 access_server 的实现了。我们将策略作为参数传递进去:

async fn access_server_v3(
  ctx : @moonchor.ChoreoContext,
  request : @moonchor.Located[Request, Client],
  strategy : ReplicationStrategy
) -> @moonchor.Located[Response, Client] {
  let request_at_server = ctx.comm(client, server, request)
  let response = strategy(ctx, request_at_server)
  ctx.comm(server, client, response)
}

async fn put_v3(
  ctx : @moonchor.ChoreoContext,
  strategy : ReplicationStrategy,
  key : String,
  value : Int
) -> Unit {
  let request = ctx.locally(client, _unwrapper => Request::Put(key, value))
  access_server_v3(ctx, request, strategy) |> ignore
}

async fn get_v3(
  ctx : @moonchor.ChoreoContext,
  strategy : ReplicationStrategy,
  key : String
) -> @moonchor.Located[Response, Client] {
  let request = ctx.locally(client, _unwrapper => Request::Get(key))
  access_server_v3(ctx, request, strategy)
}

这样一来,复制策略被成功从处理请求的逻辑中抽象出来了。下面,我们重新实现一遍双副本的复制策略:

async fn double_replication_strategy(
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup : @moonchor.Located[ServerState, Backup],
) -> ReplicationStrategy {
  fn(
    ctx : @moonchor.ChoreoContext,
    request_at_server : @moonchor.Located[Request, Server]
  ) {
    let request_at_backup = ctx.comm(server, backup, request_at_server)
    let response_at_backup = ctx.locally(backup, fn(unwrapper) {
      let request = unwrapper.unwrap(request_at_backup)
      let state = unwrapper.unwrap(state_at_backup)
      handle_request(state, request)
    })
    let backup_response = ctx.comm(backup, server, response_at_backup)
    ctx.locally(server, fn(unwrapper) {
      let request = unwrapper.unwrap(request_at_server)
      let state = unwrapper.unwrap(state_at_server)
      let res = handle_request(state, request)
      check_consistency([unwrapper.unwrap(backup_response), res])
      res
    })
  }
}

注意看 double_replication_strategy 的函数签名,它返回一个 ReplicationStrategy 类型的函数。只要提供两个参数,double_replication_strategy 就能构造出一个新的复制策略。至此,我们成功利用高阶函数抽象出了复制策略,这个特性在协同式编程中叫作高阶 choreography。

同样的,我们可以写一个简单的 choreography 来测试它:

async fn kvstore_v3(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  let state_at_backup = ctx.locally(backup, _unwrapper => ServerState::new())
  let strategy = double_replication_strategy(state_at_server, state_at_backup)
  put_v3(ctx, strategy, "key1", 42)
  put_v3(ctx, strategy, "key2", 41)
  let v1_at_client = get_v3(ctx, strategy, "key1")
  let v2_at_client = get_v3(ctx, strategy, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore 3.0" {
  let backend = @moonchor.make_local_backend([server, client, backup])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, server))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, client))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v2, backup))
}

利用参数化多态实现角色多态​

如果要进一步实现新的复制策略,例如三副本,我们需要定义两个新的 Backup 类型以做区分:

struct Backup1 {} derive(Eq, Hash, Show)

impl @moonchor.Location for Backup1 with name(_) {
  "backup1"
}

let backup1 : Backup1 = Backup1::{  }

struct Backup2 {} derive(Eq, Hash, Show)

impl @moonchor.Location for Backup2 with name(_) {
  "backup2"
}

let backup2 : Backup2 = Backup2::{  }

接下来需要修改 access_server 的核心逻辑。我们立刻发现了问题,为了让 Backup1 和 Backup2 都处理一遍请求并且得到响应,需要将以下几条语句重复:let request = unwrapper.unwrap(request_at_backup); let state = unwrapper.unwrap(state_at_backup); handle_request(state, request)。重复代码是坏味道,应当被抽象出来。此时,moonchor 的「角色作为类型」优势就体现出来了,我们可以利用 MoonBit 的参数化多态,将从副本处理逻辑抽象成一个多态函数 do_backup,它接收一个角色类型参数 B,表示从副本的角色:

async fn[B : @moonchor.Location] do_backup(
  ctx : @moonchor.ChoreoContext,
  request_at_server : @moonchor.Located[Request, Server],
  backup : B,
  state_at_backup : @moonchor.Located[ServerState, B]
) -> @moonchor.Located[Response, Server] {
  let request_at_backup = ctx.comm(server, backup, request_at_server)
  let response_at_backup = ctx.locally(backup, fn(unwrapper) {
    let request = unwrapper.unwrap(request_at_backup)
    let state = unwrapper.unwrap(state_at_backup)
    handle_request(state, request)
  })
  ctx.comm(backup, server, response_at_backup)
}

如此一来,我们就能随心所欲地实现双副本或者三副本的复制策略了。对于三副本策略,只需在 triple_replication_strategy 返回的函数内调用 do_backup 两次即可:

async fn triple_replication_strategy(
  state_at_server : @moonchor.Located[ServerState, Server],
  state_at_backup1 : @moonchor.Located[ServerState, Backup1],
  state_at_backup2 : @moonchor.Located[ServerState, Backup2]
) -> ReplicationStrategy {
  fn(
    ctx : @moonchor.ChoreoContext,
    request_at_server : @moonchor.Located[Request, Server]
  ) {
    let backup_response1 = do_backup(
      ctx, request_at_server, backup1, state_at_backup1,
    )
    let backup_response2 = do_backup(
      ctx, request_at_server, backup2, state_at_backup2,
    )
    ctx.locally(server, fn(unwrapper) {
      let request = unwrapper.unwrap(request_at_server)
      let state = unwrapper.unwrap(state_at_server)
      let res = handle_request(state, request)
      check_consistency([
        unwrapper.unwrap(backup_response1),
        unwrapper.unwrap(backup_response2),
        res,
      ])
      res
    })
  }
}

由于我们成功完成了复制策略和访问过程的分离,access_server、put、get 函数不需要任何修改。让我们对最终的 KVStore 进行测试:

async fn kvstore_v4(ctx : @moonchor.ChoreoContext) -> Unit {
  let state_at_server = ctx.locally(server, _unwrapper => ServerState::new())
  let state_at_backup1 = ctx.locally(backup1, _unwrapper => ServerState::new())
  let state_at_backup2 = ctx.locally(backup2, _unwrapper => ServerState::new())
  let strategy = triple_replication_strategy(
    state_at_server, state_at_backup1, state_at_backup2,
  )
  put_v3(ctx, strategy, "key1", 42)
  put_v3(ctx, strategy, "key2", 41)
  let v1_at_client = get_v3(ctx, strategy, "key1")
  let v2_at_client = get_v3(ctx, strategy, "key2")
  ctx.locally(client, fn(unwrapper) {
    let v1 = unwrapper.unwrap(v1_at_client).unwrap()
    let v2 = unwrapper.unwrap(v2_at_client).unwrap()
    if v1 + v2 == 83 {
      println("The server is working correctly")
    } else {
      panic()
    }
  })
  |> ignore
}

test "kvstore 4.0" {
  let backend = @moonchor.make_local_backend([server, client, backup1, backup2])
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, server))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, client))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, backup1))
  @toolkit.run_async(() => @moonchor.run_choreo(backend, kvstore_v4, backup2))
}

至此,我们完成了多副本 KVStore 的构建。在这个例子中,我们没有手动使用任何 send 和 recv 来表达分布式节点间的交互,而是通过 moonchor 的协同式编程能力实现了所有通信和同步过程,避免可能的类型错误、死锁和显式同步问题。

结语​

在这篇文章中,我们借助 moonchor 体验了协同式编程的魅力,还见识了 MoonBit 强大的表达能力。关于协同式编程的更多细节,可以参考 Haskell 的库 HasChor、Choral 语言、moonchor 的源码。想要自己尝试使用 moonchor,可以通过 moon add Milky2018/moonchor@0.15.0 命令安装。

MoonBit Pearls Vol.03:01背包问题

· 阅读需 13 分钟

01背包问题是算法竞赛中经典的dp题目。文中总共包含五个版本的代码。从最朴素的枚举法开始,在不断的改进下,最终变成了dp解法。

问题定义​

有若干个物品,每件物品的有重量weight和价值value:

struct Item {
  weight : Int
  value : Int
}

现在,给定一个物品列表items,和背包的容量capacity。从中选出若干件物品,使得这些物品的总重量不超过背包的容量,且物品的总价值最大。

typealias @list.T as List

let items_1 : List[Item] = @list.of([
  { weight: 7, value: 20 },
  { weight: 4, value: 10 },
  { weight: 5, value: 11 },
])

以上面的items_1为例,假设背包容量是1010,那么最优的方案是选取后两个物品,占用4+5=94+5=9的容量,总共有10+11=2110+11=21点价值。

注意,由于我们不能把物品切割,因此优先挑选性价比最高的物品并非正解。例如,在上面的例子中,若选取了性价比最高的物品1,则只有2020点价值,而此时背包已经放不下其他物品了。

问题建模​

我们先定义一些基础的对象与操作。

//物品的组合,下文简称组合
struct Combination {
  items : List[Item]
  total_weight : Int
  total_value : Int
}

//空的组合
let empty_combination : Combination = {
  items: @list.empty(),
  total_weight: 0,
  total_value: 0,
}

//往组合中添加物品,得到新的组合
fn Combination::add(self : Combination, item : Item) -> Combination {
  {
    items: self.items.add(item),
    total_weight: self.total_weight + item.weight,
    total_value: self.total_value + item.value,
  }
}

//两个组合等效,意思是它们总价值一样
impl Eq for Combination with op_equal(self, other) {
  self.total_value == other.total_value
}

//比较两个组合的大小,就是比较它们总价值的大小
impl Compare for Combination with compare(self, other) {
  self.total_value.compare(other.total_value)
}

然后,我们就可以开始思考如何解决问题了。

一、朴素的枚举​

枚举法是最朴素的方案,我们依照问题的定义,一步一步执行,就能得到答案:

  1. 枚举出所有的组合;
  2. 过滤出其中有效的组合,也就是那些能装入背包的;
  3. 答案是其中总价值最大的那个。

得益于标准库提供的两个函数,我们可以将上面三行文字一比一地翻译为MoonBit代码。其中all_combinations是我们后续需要实现的函数,它的类型是(List[Item]) -> List[Combination]。

fn solve_v1(items : List[Item], capacity : Int) -> Combination {
  all_combinations(items)
  .filter(fn(comb) { comb.total_weight <= capacity })
  .unsafe_maximum()
}

注意这里使用的是unsafe_maximum而不是maximum。这是因为空列表列表中没有最大值,maximum在这种情况下会返回一个None。但我们知道,题目保证答案存在(只要capacity不是负数),所以我们可以改用unsafe_maximum。它在输入空列表的情况下直接中断程序,其它情况返回列表中的最大值。

接下来我们去实现枚举的过程。函数all_combinations接受一个物品的列表,返回一个组合的列表,其中包含所有能由这些物品构造出的组合。也许你现在没有头绪,这时我们可以先查看一下列表的定义。它大概长这样:

enum List[A] {
  Empty
  More(A, tail~ : List[A])
}

也就是说,列表分为两种:

  1. 第一种是空的列表,叫Empty;
  2. 第二种是非空的列表,叫More,其中包含了第一个元素(A)和剩余的部分(tail~ : T[A]),剩余部分也是一个列表。

这启示我们按物品列表是否为空来分情况讨论:

  • 如果物品列表为空,那么唯一的一种组合方式就是空的组合;
  • 否则,一定存在第一个物品item1和剩余部分items_tail。这种情况下,我们可以:
    1. 先求出不含item1的那些组合。这其实就是items_tail能凑出的那些组合,可以递归地求出。
    2. 再求出包含item1的那些组合。它们与不含item1的组合一一对应,只差把item1加入其中。
    3. 将这两者合并起来,就是所有items能凑出的组合。

例如,当物品列表包含a,b,c三个元素时,答案分为以下两个部分:

不含a的部分包含a的部分
{ }{ a }
{ b }{ a, b }
{ c }{ a, c }
{ b, c }{ a, b, c }
fn all_combinations(items : List[Item]) -> List[Combination] {
  match items {
    Empty => @list.singleton(empty_combination)
    More(item1, tail=items_tail) => {
      let combs_without_item1 = all_combinations(items_tail)
      let combs_with_item1 = combs_without_item1.map(_.add(item1))
      combs_with_item1 + combs_without_item1
    }
  }
}

通过使用模式匹配(match),我们再一次将上面的五行文字一比一地翻译成了MoonBit代码。

二、提前过滤,仅枚举有效的组合​

在第一个版本中,枚举所有组合和过滤出能放入背包的组合是不相干的两个过程。在枚举的过程中,出现了很多无效的组合。这些组合早已放不进背包中,却还在后续的过程中被添加物品。不如早一点过滤它们,避免在它之上不断产生新的无效组合。观察代码,发现无效的组合只会在.map(_.add(item1))这一步产生。于是我们可以做出改进:仅向能再装下item1的组合添加item1。

我们将all_combinations改为all_combinations_valid,仅返回能装入这个背包的组合。现在枚举和过滤将交替进行。

fn all_combinations_valid(
  items : List[Item],
  capacity : Int // 添加一个参数,因为过滤需要知道背包的容量
) -> List[Combination] {
  match items {
    Empty => @list.singleton(empty_combination) // 空的组合自然是有效的
    More(item1, tail=items_tail) => {
      // 我们假设 all_combinations_valid 返回的组合都是有效的(归纳假设)
      let valid_combs_without_item1 = all_combinations_valid(
        items_tail, capacity,
      )
      // 由于添加了过滤,所以它里面的组合都是有效的
      let valid_combs_with_item1 = valid_combs_without_item1
        .filter(fn(comb) { comb.total_weight + item1.weight <= capacity })
        .map(_.add(item1))
      // 两个部分都仅包含有效组合,所以合并后也仅包含有效组合
      valid_combs_with_item1 + valid_combs_without_item1
    }
  }
}

遵循代码的结构进行分类讨论,很容易证明all_combinations_valid的正确性——它返回的所有组合确实都是有效的。

由于all_combinations_valid返回的那些组合都是有效的,就不再需要在solve中过滤了。我们将solve中的filter删去。

fn solve_v2(items : List[Item], capacity : Int) -> Combination {
  all_combinations_valid(items, capacity).unsafe_maximum()
}

三、维护升序性质,提前结束过滤​

在上个版本中,为了过滤出那些能装下item1的组合,我们必须遍历valid_combs_without_item1中的每一个组合。

但我们可以发现:如果item1没法放入一个组合,那么item1一定都无法放入比这个组合总重量更大的那些组合。

这也就是说,如果valid_combs_without_item1能按总重量升序排列,那么过滤时就不需要完整地遍历它了。在过滤的过程中,一旦碰到一个放不下item1的组合,就可以立刻舍去后续的所有组合。由于这种逻辑很常见,标准库提供了一个叫take_while的函数,我们用它替换掉filter。

要想让valid_combs_without_item1升序排列,可以用排序算法,但这却要遍历整个列表,违背了初衷。因此,我们得采用另一种方案:想办法让all_combinations_valid返回的列表是升序的。这需要一次递归的信仰之跃:

fn all_combinations_valid_ordered(
  items : List[Item],
  capacity : Int
) -> List[Combination] {
  match items {
    Empty => @list.singleton(empty_combination) // 单元素的列表,自然是升序的
    More(item1, tail=items_tail) => {
      // 我们假设 all_combinations_valid_ordered 返回的列表是升序的(归纳假设)
      let valid_combs_without_item1 = all_combinations_valid_ordered(
        items_tail, capacity,
      )
      // 那么它也是升序的,因为一个升序的列表先截取一部分,再往每个元素加上同样的重量,它们的总重量还是升序的
      let valid_combs_with_item1 = valid_combs_without_item1
        .take_while(fn(comb) { comb.total_weight + item1.weight <= capacity })
        .map(_.add(item1))
      // 现在我们只需要确保合并后也升序,就能衔接上最开始的假设
      merge_keep_order(valid_combs_with_item1, valid_combs_without_item1)
    }
  }
}

最后的任务是完成函数merge_keep_order,它将两个升序的列表合并为一个升序的列表:

fn merge_keep_order(
  a : List[Combination],
  b : List[Combination]
) -> List[Combination] {
  match (a, b) {
    (Empty, another) | (another, Empty) => another
    (More(a1, tail=a_tail), More(b1, tail=b_tail)) =>
      // 如果 a1 比 b1 更轻,而 b 又是升序的,说明
      //   a1 比 b 里所有组合都轻
      // 由于 a 是升序的,所以
      //   a1 比 a_tail 里所有组合都轻
      // 所以 a1 是 a 和 b 中最小的那一个
      if a1.total_weight < b1.total_weight {
        // 我们先递归地合并出答案的剩余部分,再把 a1 加到开头
        merge_keep_order(a_tail, b).add(a1)
      } else { // 同理
        merge_keep_order(a, b_tail).add(b1)
      }
  }
}

虽然看起来有点啰嗦,但我还是想提一句:通过遵循代码结构的分类讨论,很容易证明all_combinations_valid_ordered和merge_keep_order的正确性——它确实返回的一个升序的列表。

对于一个升序的列表,它的最大值就是最后一个。于是我们将unsafe_maximum替换成unsafe_last。

fn solve_v3(items : List[Item], capacity : Int) -> Combination {
  all_combinations_valid_ordered(items, capacity).unsafe_last()
}

回过头来看,在这一版的改进中,我们似乎并没有得到什么太大的好处,毕竟在合并列表的过程中,我们仍然需要遍历整个列表。最初我也是这么想的,但后来意外地发现merge_keep_order的真正作用在下一个版本。

四、去除等同重量的冗余组合,达到最优时间复杂度​

目前为止,我们进行的都不是时间复杂度层面的优化,但这些优化恰恰为接下来的步骤铺平了道路。现在让我们来考察一下时间复杂度。

在最差情况下(背包很大,全都放得下),组合列表(all_combinations的返回值)将最多包含 2物品数量2^{物品数量} 个元素。这导致整个算法的时间复杂度也是指数级的,因为all_combinations会被调用 物品数量物品数量 次,而每次都会遍历组合列表。

为了降低时间复杂度,我们就需要降低组合列表的长度。这基于一个观察:如果有两个组合,它们总重量相同,那么总价值更高的那个组合总是比另一个更好。因此,我们不需要在列表中同时保留两者。

如果能排除那些冗余的组合,组合列表的长度将不会超过背包容量(抽屉原理),进而将整个算法的时间复杂度降低到 O(物品数量×背包容量)\mathcal{O}(物品数量 \times 背包容量)。观察代码,现在唯一有可能会向列表中引入冗余组合的地方是merge_keep_order的else分支。为了避免这种情况出现,我们只需要对这个地方进行一点改动:

fnalias @math.maximum

fn merge_keep_order_and_dedup(
  a : List[Combination],
  b : List[Combination]
) -> List[Combination] {
  match (a, b) {
    (Empty, another) | (another, Empty) => another
    (More(a1, tail=a_tail), More(b1, tail=b_tail)) =>
      if a1.total_weight < b1.total_weight {
        merge_keep_order_and_dedup(a_tail, b).add(a1)
      } else if a1.total_weight > b1.total_weight {
        merge_keep_order_and_dedup(a, b_tail).add(b1)
      } else { // 此时 a1 和 b1 一样重,出现冗余,保留总价值更高的那个
        let better = maximum(a1, b1)
        merge_keep_order_and_dedup(a_tail, b_tail).add(better)
      }
  }
}

all_combinations_valid_ordered_nodup(这是我这辈子写过的名字最长的函数了)和solve_v4替换相应部分即可。

fn all_combinations_valid_ordered_nodup(
  items : List[Item],
  capacity : Int
) -> List[Combination] {
  match items {
    Empty => @list.singleton(empty_combination)
    More(item1, tail=items_tail) => {
      let combs_without_item1 = all_combinations_valid_ordered_nodup(
        items_tail, capacity,
      )
      let combs_with_item1 = combs_without_item1
        .take_while(fn(comb) { comb.total_weight + item1.weight <= capacity })
        .map(_.add(item1))
      merge_keep_order_and_dedup(combs_with_item1, combs_without_item1)
    }
  }
}

fn solve_v4(items : List[Item], capacity : Int) -> Combination {
  all_combinations_valid_ordered_nodup(items, capacity).unsafe_last()
}

至此,我们重新发明了01背包问题的dp解法。

总结​

这篇文章的内容是我某天早上躺在床上的突发奇想,从第一版到第四版代码完全在手机上写成,没有经过任何调试,但却能轻松地保证了正确性。相比传统算法竞赛题解中常见的写法,本文中使用的函数式写法带来了以下优势:

  1. 告别循环,使用递归分情况讨论。要想从列表中获取元素,必须使用模式匹配(match),这提醒我考虑列表为空时的答案。它相比dp数组的初始值拥有更加明确的含义。
  2. 依赖库函数进行遍历。标准库中提供的高阶函数(filter、take_while、map、maximum)能替换掉样板化的循环(for、while),便于读者一眼看出遍历的目的。
  3. 声明式编程。第一版的代码是想法的一比一地翻译。与其说是在描述一个算法,更像是在描述这个问题本身,这保证了第一版的正确性。而随后每次改进都在不影响结果的前提下进行,于是继承了第一版的正确性。

当然,从来就没有银弹。我们需要可读性和效率之间做取舍。函数式的风格固然好理解,但还是有许多优化余地的。进一步的优化方向是将列表替换成数组,再替换成从头到尾只使用两个滚动数组,甚至是只使用一个数组。这可以将空间复杂度优化成 O(背包容量)\mathcal{O}(背包容量),但不在本文的讨论范围内。我相信初学者更希望看到的是一个易于理解的代码。

附录​

更多细节优化​

利用items中物品的顺序不影响结果的总价值这个性质。可以把all_combinations转化成尾递归。

另外,take_while产生的列表在map后马上就被丢弃了,我们可以改用迭代器来避免产生这个一次性列表。

fn all_combinations_loop(
  items : List[Item],
  capacity : Int
) -> List[Combination] {
  loop items, @list.singleton(empty_combination) {
    Empty, combs_so_far => combs_so_far
    More(item1, tail=items_tail), combs_so_far => {
      let combs_without_item1 = combs_so_far
      let combs_with_item1 = combs_without_item1
        .iter()
        .take_while(fn(comb) { comb.total_weight + item1.weight <= capacity })
        .map(_.add(item1))
        |> @list.from_iter
      continue items_tail,
        merge_keep_order_and_dedup(combs_with_item1, combs_without_item1)
    }
  }
}

fn solve_v5(items : List[Item], capacity : Int) -> Combination {
  all_combinations_loop(items, capacity).unsafe_last()
}

题外话​

  1. 在第一版中,all_combinations(items)产生的Combination甚至比其中的More还多一个,堪称链表节点复用大师。
  2. 升序还可以换成降序,对应的take_while要换成drop_while。而改用Array后可以通过binary_search来寻找下标直接切分。
  3. 如果你感兴趣,可以考虑一下怎么把上面的做法拓展到各种其它的背包问题。
  4. all_combinations_loop原名:generate_all_ordered_combination_that_fit_in_backpack_list_without_duplicates_using_loop。

测试​

test {
  for solve in [solve_v1, solve_v2, solve_v3, solve_v4, solve_v5] {
    assert_eq(solve(items_1, 10).total_value, 21)
  }
}