基于列表的数据流编程

数据流编程范式

数据流编程是一种编程范式，其中我们将程序建模为由节点和连接节点的有向边组成的有向图。一个节点代表一个接受输入并产生输出的操作。从节点A到节点B的有向边将A的输出作为B的输入。

为了说明这一点，我们以伯特·萨瑟兰（Bert Sutherland）博士论文《在线图形化计算机过程规范》中的算术计算为例，该论文开创了数据流编程。

与文本表示相比，图形化表示通过强调数据如何从一个操作流向另一个操作，体现了数据流编程风格。这具有多个优势。首先，数据流导向的程序描述了程序由更小的构建模块组成，因此相对容易理解。其次，数据流版本允许无需程序员额外努力即可实现并行执行。数据流程序的执行器可以自动根据操作之间的数据依赖关系判断哪些操作可以并行运行。

在我们的例子中，W 和 Y 可以同时计算，因为它们互不依赖。

这种范式通常鼓励创建通用组件库。这个库可用于构建程序，通常以可视化编程的形式呈现。当我们混合搭配现有组件来构建程序时，相比从零开始编写和实现，能节省大量开发时间和精力。此外，这些构建模块可以组合起来解决其创建者甚至未曾设想的问题。

数据流编程常被嵌入式软件工程师用于开发控制系统，例如汽车巡航控制。函数式编程语言使我们能够以数据流导向的方式编程，并享受其全部优势。

函数作为数据流组件

在函数式编程语言中，函数是纯函数。这意味着对于相同的输入，它们总是返回相同的结果，且调用不会产生任何副作用。由于这些特性，函数式范式中的函数行为类似于数据流组件——它们也接受输入并产生输出，且不会引起任何副作用。

注意：如果一组函数在输入/输出数据格式上达成一致，我们可以以数据流方式组合它们。

让我们考虑三个函数——map、filter 和 fold。我们可以将这些函数视为数据流组件。特别是，map 作为一个数据流组件，将输入信号的每个元素映射为输出信号的新值。它是一个通用组件，适用于多种场景，因为它允许我们配置映射函数 f 以满足任意需求。

filter 组件仅允许满足谓词条件的输入信号元素通过。它也是一个高度可重用的组件，因为它允许我们配置谓词以适应特定用途。

我们也可以将 fold 视为一个数据流组件，它将输入信号累积为一个值。我们可以配置其初始值和累积函数。

所有这些函数都使用列表作为共同接口。例如，map 和 filter 接受列表作为输入并返回列表作为输出。fold 函数接收一个列表作为参数并返回聚合值。这使得我们可以构造以 map、filter、fold 等列表处理函数为节点的数据流程序图。

为了说明，假设我们要实现一个函数 sum_even_squares，用于计算区间 a 到 b 内所有偶数的平方和。例如： $sum\_even\_squares \; 1\; 4 = 2^2 + 4^2 = 20$

当然，我们可以快速编写一个递归函数来完成此任务，如下所示：

1(** [even x] 在 [x] 为偶数时返回 [true]，否则返回 [false]。 *)
2let even x = x mod 2 = 0
3
4(** [square x] 是 [x] 的平方。 *)
5let square x = x * x
6
7(** [sum_even_squares a b] 是 [a] 到 [b] 之间偶数的平方和。 *)
8let rec sum_even_squares a b =
9   if a > b then 0 else
10   let x = if even a then square a else 0 in x + sum_even_squares (a + 1) b
11
12(** 将 'sum_even_squares 1 4' 的结果打印到控制台 *)
13let _ = print_int (sum_even_squares 1 4)

然而，无需从头实现 sum_even_squares。相反，更高效的做法是复用现有的组件——map、filter 和 fold 来组合解决方案。关键在于认识到我们可以将 sum_even_squares 定义为一个数据流程序，按以下阶段处理数据：

• 将 a 到 b 之间的整数枚举为一个列表。
• 过滤列表，仅保留偶数。
• 将每个元素映射为其平方。
• 使用 + 对列表元素进行折叠，从 0 开始累加求和。

下图可视化了该程序：

从图中可以看出，我们唯一需要做的就是枚举 [a, b] 区间内的整数。我们可以使用 filter、map 和 fold 来定义其余所有阶段。

我们可以定义以下函数，将 a 到 b 之间的整数枚举为列表：

1let rec enumerate_integers a b = if a > b then [] else a :: enumerate_integers (a + 1) b

接下来，我们将 sum_even_squares 的数据流图转换为 OCaml 代码，如下所示：

1(** [enumerate_integers a b] 是从 [a] 到 [b] 的整数列表。 *)
2let rec enumerate_integers a b = if a > b then [] else a :: enumerate_integers (a + 1) b
3
4(** [even x] 在 [x] 为偶数时返回 [true]，否则返回 [false]。 *)
5let even x = x mod 2 = 0
6
7(** [square x] 是 [x] 的平方。 *)
8let square x = x * x
9
10(** [sum_even_squares a b] 是 [a] 到 [b] 之间偶数的平方和。 *)
11let sum_even_squares a b = List.fold_left (+) 0 (List.map square (List.filter even (enumerate_integers a b)))
12
13(** 将 'sum_even_squares 1 4' 的结果打印到控制台。 *)
14let _ = print_int (sum_even_squares 1 4)

有一点不够优化。虽然数据流图显示了从左到右的处理阶段，但文本版本顺序相反。我们可以使用 OCaml 的管道操作符 |> 来改变这一顺序。它允许我们从左到右将参数传递给函数。该操作符定义如下：

1x |> f = f x

我们可以使用此操作符创建 sum_even_squares 的改进版本。

1let sum_even_squares a b = enumerate_integers a b |> List.filter even |> List.map square |> List.fold_left (+) 0

在上方的代码沙盒中尝试这个改进版本。

该版本更接近上述数据流图，因为它强调数据从左到右流动。

作为另一个例子，假设我们要编写另一个函数 sum_tree_even_squares。该函数对二叉树中所有偶数节点的值求平方和。二叉树由以下代数数据类型表示：

1type 'a bin_tree =
2        | Leaf
3        | Node of 'a bin_tree * 'a * 'a bin_tree

采用数据流编程范式，我们可以构建以下数据流图来解决问题：

除了第一阶段外，这与上面针对列表的 sum_even_squares 完全相同。我们需要将树中的所有元素枚举为列表。可以通过对树进行折叠来实现，如下所示：

1let enumerate_tree_elements l = fold_tree (fun l x r -> x :: l @ r) [] l

这里 fold_tree 用于折叠二叉树。

1let rec fold_tree f init t = match t with
2                             | Leaf -> init
3                             | Node (l, x, r) -> f (fold_tree f init l) x (fold_tree f init r)

接下来，我们将组件连接起来，形成 sum_tree_even_squares。

1(** ['a bin_tree] 数据类型定义二叉树。 *)
2type 'a bin_tree =
3   | Leaf
4   | Node of 'a bin_tree * 'a * 'a bin_tree
5
6(** [fold_tree f init t] 使用函数 [f] 从初始值 [init] 开始折叠树 [t]。 *)
7let rec fold_tree f init t = 
8   match t with
9   | Leaf -> init
10   | Node (l, x, r) -> f (fold_tree f init l) x (fold_tree f init r)
11
12(** [enumerate_tree_elements t] 是包含树 [t] 所有元素的列表。 *)
13let enumerate_tree_elements t = fold_tree (fun l x r -> x :: l @ r) [] t
14
15(** [even x] 在 [x] 为偶数时返回 [true]，否则返回 [false]。 *)
16let even x = x mod 2 = 0
17
18(** [square x] 是 [x] 的平方。 *)
19let square x = x * x
20
21(** [sum_tree_even_squares t] 是树 [t] 中元素的平方和。 *)
22let sum_tree_even_squares t = enumerate_tree_elements t |> List.filter even |> List.map square |> List.fold_left (+) 0
23
24(** 将 'sum_tree_even_squares (Node (Node (Leaf, 2, Leaf), 1, Node (Node (Leaf, 4, Leaf), 3, Leaf)))' 的结果打印到控制台 *)
25let _ = print_int (sum_tree_even_squares (Node (Node (Leaf, 2, Leaf), 1, Node (Node (Leaf, 4, Leaf), 3, Leaf))))

函数 map 和 fold 接受单个列表作为输入。因此，从数据流编程的角度看，我们可以将它们视为单一输入信号的组件。对于具有多个输入列表的函数，我们可以将其视为接受多个输入信号的数据流组件。例如，zipWith 作为一个组件，从两个输入信号生成一个输出信号。

让我们使用 zipWith 构建一个数据流程序。该程序将以字符串列表的形式统计小于或等于 n 的质数数量。例如，若 n 为 6，则输出应为 ["Prime at 1 is 2"; "Prime at 2 is 3", "Prime at 3 is 5"]。

以下是解决此问题的一种数据流图：

我们可以将该图转换为 OCaml 代码，如下所示。

1(** [print_string_list] 将字符串列表 [l] 打印到控制台。 *)
2let print_string_list l = print_string ("[" ^ (String.concat "; " l) ^ "]")
3
4(** [zipWith f l1 l2] 将两个列表 [l1] 和 [l2] 打包成一对列表。 *)
5let rec zipWith f l1 l2 = 
6   match l1, l2 with
7   | [], _ -> []
8   | _, [] -> []
9   | x :: xs, y :: ys -> f x y :: zipWith f xs ys
10
11(** [is_prime n] 在 [n] 为质数时返回 [true]，否则返回 [false]。 *)
12let rec is_prime n =
13    if n < 2 then false else if n = 2 then true else
14        let rec aux m = if n mod m = 0 then false
15                        else if m * m > n then true else aux (m + 1) in aux 2
16
17(** [enumerate_integers a b] 是从 [a] 到 [b] 的整数列表。 *)
18let rec enumerate_integers a b =
19   if a > b then [] else a :: enumerate_integers (a + 1) b
20
21(** [count_primes n] 是模拟从 1 到 [n] 之间计数质数的一组字符串列表。 *)
22let count_primes n =
23   let l = enumerate_integers 1 n in
24   zipWith (fun x y -> "Prime at " ^ (string_of_int x) ^ " is " ^ (string_of_int y)) l (List.filter is_prime l)
25
26let _ = print_string_list (count_primes 6)

组合规则

在埃里克·斯蒂芬·雷蒙德（Eric Steven Raymond）的精彩著作《Unix编程艺术》中，他讨论了 Unix 哲学——基于成功开发 Unix 操作系统的经验总结出的一套软件开发反思。

Unix 哲学的一个核心方面是雷蒙德所称的 组合规则。它指出，我们应该优先选择小型、独立的程序，这些程序只做一件事，并且做好这件事。此外，我们应致力于构建输入和输出为纯文本流而非二进制数据的程序。纯文本充当程序间交换信息的通用接口。得益于这一点，一个程序的输出可以通过管道操作符（竖线 |）直接作为另一个程序的输入。

例如，Unix 或 macOS 系统通常提供以下程序：

• cat：用于创建文件或查看文件内容。
• head：打印输入的前几行。例如，head -5 打印前 5 行。
• tail：打印输入的后几行。例如，tail -5 打印最后 5 行。

在 Unix 或 macOS 系统上，/usr/share/dict/words 文件是一个英文词典，包含每行一个单词：

1A
2a
3aa
4aal
5aalii
6aam
7Aani
8aardvark
9aardwolf
10Aaron
11Aaronic
12...

如何打印词典中的第 5 个单词？无需从头编写新程序，我们可以组合已有的程序 cat、head 和 tail 实现此功能。我们使用以下命令完成：

1cat /usr/share/dict/words | head -5 | tail -1
2aalii

下图展示了程序之间的连接方式：

本质上，cat 读取词典内容，通过管道操作符 | 发送给 head。head -5 程序返回前 5 行。然后，tail -1 程序从中选取最后一行，即第 5 个单词。

在整个课程中，我们始终遵循组合规则。特别是，map、filter、fold 和 zipWith 这些函数是小型、独立的程序，能够很好地完成单一任务。例如，map 擅长元素变换，而 fold 适合数据聚合。使这些程序可组合的通用接口是 list。OCaml 的管道操作符 |> 在概念上类似于 Unix 的管道操作符 |。组合规则是 Unix 操作系统如此成功的原因之一。我们可以在函数式编程中利用同样的力量。