20.1 Parallel Threads🔗

使用 并行线程池创建的 线程可以使用与其他线程不同的硬件处理器，从而在并发之外提供并行性。当硬件可用时，使用这些 并行线程可以减少计算的延迟。

并行线程的一个权衡是，由于创建和管理并行任务的开销，计算可能变慢而不是变快。

Racket 的 BC 实现将所有线程视为 协程线程，并且不并行运行它们。

使用 thread 的 #:pool 参数创建并行线程，其中 'own 表示该线程拥有自己的单线程池。

(define (fib n)

(cond

[(= n 0) 1]

[(= n 1) 1]

[else (+ (fib (- n 1)) (fib (- n 2)))]))

(define n1 (thread (lambda () (fib 35))

#:pool 'own

#:keep 'results))

(define n2 (thread (lambda () (fib 35))

#:pool 'own

#:keep 'results))

(time (= (thread-wait n1)

(thread-wait n2)))

并行线程是否能提高性能取决于计算的性质。可以高效并行运行的计算类型大多与可以在 future 中运行的类型相同，Parallelism with Futures 提供了更多信息并描述了 future 可视化器如何帮助解释程序的性能。与并行线程的区别在于，当 future 会阻塞时，并行线程反而与其他线程同步以继续——希望很快达到一个没有阻塞动作的清晰点。此外，并行线程可以自由使用 参数和异常处理器，这些在 future 中受限，直到 touch 为 future 提供完整的动态上下文。最后，并行线程实现了与 协程线程相同的同步接口，包括 thread-wait、thread-send 和 kill-thread。

Racket 的预定义构造（如输入和输出端口或可变哈希表）是线程安全的，并在内部按需要使用锁。线程可以并行执行一些输入和输出操作，特别是从不会阻塞的常规文件。同时，阻塞的输入和输出操作不仅干扰并行性，而且对于并行线程来说，它们比协程线程更昂贵得多，因为需要额外的调度。与线程调度交互的操作（如 sleep、sync、thread、thread-wait、kill-thread 和 custodian-shutdown-all）都需要调度，当从并行线程触发时，这些调度可能更昂贵。

semaphore-post 和 semaphore-wait 操作在并行线程中可以很便宜，只要没有线程最终阻塞在信号量上。当线程阻塞在信号量上时，对信号量操作与线程调度相互作用，因此在并行线程中更慢。在某些情况下，使用 future 信号量（比普通信号量更有限）可以为并行线程提供更好的性能。

20.2 Parallelism with Futures🔗

racket/future 库提供了通过 futures 实现并行性能提升的支持， 主要通过 future 和 touch 函数来实现。 通过这些构造实现并行依赖于避免 blocking 操作——即任何检查完整 continuation 或相对于 Racket threads 需要原子执行的操作。 Performance in DrRacket 中的注意事项同样适用于 futures；特别是， 调试检测目前会使 futures 失效。

其他函数，如 thread，支持创建可靠并发的任务。

作为入门示例，下面的 any-double? 函数接收一个数字列表， 并判断列表中是否存在某个数字的两倍也在列表中：

(define (any-double? l)

(for/or ([i (in-list l)])

(for/or ([i2 (in-list l)])

(= i2 (* 2 i)))))

该函数以二次时间运行，因此在 l1 和 l2 等 大型列表上可能需要较长时间（大约一秒）：

(define l1 (for/list ([i (in-range 5000)])

(+ (* 2 i) 1)))

(define l2 (for/list ([i (in-range 5000)])

(- (* 2 i) 1)))

(or (any-double? l1)

(any-double? l2))

加速 any-double? 的最佳方式是使用不同的算法。 然而，在提供至少两个处理单元的机器上， 上述示例可以使用 future 和 touch 在大约一半的时间内运行：

(let ([f (future (lambda () (any-double? l2)))])

(or (any-double? l1)

(touch f)))

future f 与 (any-double? l1) 并行运行 (any-double? l2)，而 (any-double? l2) 的结果 大约在 (touch f) 需要它时就已可用。

Futures 只要能够安全地运行且独立于 touch 可能提供的 任何 continuation 上下文，就会并行运行。Continuation 上下文可以包括 异常处理器、参数值、continuation prompts 或通过 continuation marks 可访问的其他值。安全性涉及共享的可变对象，如输入端口、输出端口 和基于 equal? 的哈希表，这些对象的并发访问通过内部管理 或通过锁显式管理。本节的剩余部分通过一个示例来说明安全性和 continuation 障碍，同时展示了 futures visualizer （由 future-visualizer 提供） 如何帮助揭示这些障碍。

考虑以下 Mandelbrot 集合计算的核心部分：

(define (mandelbrot iterations x y n)

(printf "starting\n")

(let ([ci (- (/ (* 2.0 y) n) 1.0)]

[cr (- (/ (* 2.0 x) n) 1.5)])

(let loop ([i 0] [zr 0.0] [zi 0.0])

(if (> i iterations)

i

(let ([zrq (* zr zr)]

[ziq (* zi zi)])

(cond

[(> (+ zrq ziq) 4) i]

[else (loop (add1 i)

(+ (- zrq ziq) cr)

(+ (* 2 zr zi) ci))]))))))

表达式 (mandelbrot 10000000 62 500 1000) 和 (mandelbrot 10000000 62 501 1000) 各自都需要 一段时间才能产生结果。当然，同时计算两者需要两倍的时间：

(list (mandelbrot 10000000 62 500 1000)

(mandelbrot 10000000 62 501 1000))

不幸的是，尝试使用一个 future 来并行运行这两个计算 并不能改善性能：

(let ([f (future (lambda () (mandelbrot 10000000 62 501 1000)))])

(list (mandelbrot 10000000 62 500 1000)

(touch f)))

要了解原因，请使用 future-visualizer 中的 futures visualizer 来可视化上述程序的执行。 可视化器会打开一个窗口，显示计算轨迹的图形视图。 窗口的左上部分包含一个执行时间线：

每一行代表一个并行任务，彩色圆点代表程序执行中的重要事件； 它们通过颜色编码来区分不同的事件类型。时间线左上角的蓝色圆点 代表 future 的创建。该 future 在 thread 1 上短暂执行 （由第二行的绿色条表示）。然后它暂停，因为 Racket thread 需要执行一个 future 不安全的操作（由红色圆点表示）。 暂停时间较长，因为 Racket thread 在 touch future 之前 正在执行自己的计算副本。同时，粉色的竖线代表垃圾回收事件， 这意味着跨并行任务的同步。

future 不安全的操作或需要 continuation 上下文的操作 是 blocking（阻塞）操作。 阻塞操作会停止 future 的求值，并且不允许它继续执行， 直到 future 被 touch。在 thread 中对 future 进行 touch 会导致其工作由该 thread 顺序求值， 该 thread 提供 continuation 上下文并可与 Racket threads 同步。在 Racket 的 BC 实现中， synchronized（同步）操作也会停止 future。 后台 thread 可以随时执行该操作，一旦完成， future 可以继续并行运行。CS 实现 可以在不停止 future 或依赖后台 thread 的情况下 执行同步操作。

当您将鼠标移到某个事件上时，可视化器会显示该事件的详细信息， 并绘制箭头连接相应 future 中的所有事件。 此图像显示了我们 future 的这些连接。

蓝色虚线将 future 中的第一个事件连接到 创建它的 future，红色线条将 future 的阻塞事件 连接到其恢复点，紫色线条连接 future 内部的相邻事件。

我们看不到并行的原因是 mandelbrot 中循环之前的 printf 操作需要查找 current-output-port 参数的值， 这取决于 touch 的求值上下文。即使通过使用 fprintf 和直接引用端口的变量来修复这个问题，写入端口也是一个阻塞操作， 因为它必须获取端口上的锁。移除 printf 可以同时避免这两个问题。

更一般地，我们可以创建 N 个 futures 来执行相同的计算， 它们将并行运行：

(define fs

(for/list ([i (in-range N)])

(future (lambda () (mandelbrot 10000000 62 500 1000)))))

(for/list ([f (in-list fs)])

(touch f))

在至少有 4 个处理单元的机器上，将 N 设为 4：

此示例中的几乎所有算术运算都会产生需要分配存储的不精确数， 这会触发频繁的垃圾回收，如密集的粉色线条所示， 实际上给整个图形提供了粉色背景。垃圾回收不一定是问题， 但由于垃圾回收需要跨并行任务的同步，有时会限制性能。

通过使用 flonum 特定的操作（参见 Fixnum and Flonum Optimizations），我们可以重写 mandelbrot 以使用更少的分配：

(define (mandelbrot iterations x y n)

(let ([ci (fl- (fl/ (* 2.0 (->fl y)) (->fl n)) 1.0)]

[cr (fl- (fl/ (* 2.0 (->fl x)) (->fl n)) 1.5)])

(let loop ([i 0] [zr 0.0] [zi 0.0])

(if (> i iterations)

i

(let ([zrq (fl* zr zr)]

[ziq (fl* zi zi)])

(cond

[(fl> (fl+ zrq ziq) 4.0) i]

[else (loop (add1 i)

(fl+ (fl- zrq ziq) cr)

(fl+ (fl* 2.0 (fl* zr zi)) ci))]))))))

此转换可以将 mandelbrot 加速约 10 倍， 即使在顺序模式下也是如此，但避免分配还允许 mandelbrot 更一致地并行运行。执行此程序 在可视化器中产生以下无粉色的结果 （与之前的图形不按比例）：

一般而言，如果一个操作需要查询 continuation （例如获取参数值），或者它与 Racket 的 thread 系统交互 （例如在输出端口或基于 equal? 的哈希表的实现中获取锁）， 那么该操作就是 blocking 的。在 Racket 的 CS 实现中， 大多数原语是非阻塞的，而 BC 实现包含更多 阻塞或 synchronized 操作。

20.3 Parallelism with Places🔗

racket/place 库提供了通过 place 形式进行并行性提高性能的支持。place 形式创建一个 place，有效地成为可以与其他 place（包括初始 place）并行运行的新 Racket 实例。Racket 语言的完整功能在每个 place 都可用，但 place 只能通过消息传递进行通信——在有限的一组值上使用 place-channel-put 和 place-channel-get 函数——这有助于确保并行计算的安全性和独立性。

作为初始示例，下面的 Racket 程序使用 place 来确定列表中是否有数字的double也在列表中：

#lang racket

(provide main)

(define (any-double? l)

(for/or ([i (in-list l)])

(for/or ([i2 (in-list l)])

(= i2 (* 2 i)))))

(define (main)

(define p

(place ch

(define l (place-channel-get ch))

(define l-double? (any-double? l))

(place-channel-put ch l-double?)))

(place-channel-put p (list 1 2 4 8))

(place-channel-get p))

place 后的标识符 ch 绑定到 place channel。place form 中其余的 body 表达式在新 place 中求值，body 表达式使用 ch 与产生新 place 的 place 进行通信。

在上述 place form 的 body 中，新 place 通过 ch 接收数字列表并将列表绑定到 l。然后它调用 any-double? 处理列表并将结果绑定到 l-double?。最后的 body 表达式将 l-double? 结果通过 ch 发送回原始 place。

在 DrRacket 中，保存并运行上述程序后，在交互窗口中求值 (main) 以创建新 place。在 DrRacket 内使用 places 时，包含 place 代码的 module 必须在执行前保存到文件中。 或者，将程序另存为 "double.rkt" 并使用命令行运行

  racket -tm double.rkt

其中 -t 标志告诉 racket 加载 double.rkt module，-m 标志调用导出的 main 函数，-tm 将两个标志组合起来。

place 形式有两个微妙的特性。首先，它将 place body 提升为匿名的模块级函数。这种提升意味着 place body 引用的任何绑定必须在模块的顶级可用。其次，place 形式在新建的 place 中 dynamic-require 封闭的 module。作为 dynamic-require 的一部分，当前 module body 在新 place 中求值。第二个特性的结果是 place 不应直接出现在 module 中或在模块顶级调用的函数中；否则，调用模块将在新 place 中调用同一模块，依此类推，触发一系列 place 创建，很快耗尽内存。

#lang racket

(provide main)

; 不要这样做！

(define p (place ch (place-channel-get ch)))

(define (indirect-place-invocation)

(define p2 (place ch (place-channel-get ch))))

; 也不要这样做！

(indirect-place-invocation)

20.4 Distributed Places🔗

racket/place/distributed 库提供分布式编程支持。

下面的示例演示了如何启动一个远程 Racket 节点实例、 在新的远程节点实例上启动远程 place，以及启动一个 监视远程节点实例的 event loop。

示例代码也可以在 "racket/distributed/examples/named/master.rkt" 中找到。

#lang racket/base

(require racket/place/distributed

racket/class

racket/place

racket/runtime-path

"bank.rkt"

"tuple.rkt")

(define-runtime-path bank-path "bank.rkt")

(define-runtime-path tuple-path "tuple.rkt")

(provide main)

(define (main)

(define remote-node (spawn-remote-racket-node

"localhost"

#:listen-port 6344))

(define tuple-place (supervise-place-at

remote-node

#:named 'tuple-server

tuple-path

'make-tuple-server))

(define bank-place  (supervise-place-at

remote-node bank-path

'make-bank))

(message-router

remote-node

(after-seconds 4

(displayln (bank-new-account bank-place 'user0))

(displayln (bank-add bank-place 'user0 10))

(displayln (bank-removeM bank-place 'user0 5)))

(after-seconds 2

(define c (connect-to-named-place remote-node

'tuple-server))

(define d (connect-to-named-place remote-node

'tuple-server))

(tuple-server-hello c)

(tuple-server-hello d)

(displayln (tuple-server-set c "user0" 100))

(displayln (tuple-server-set d "user2" 200))

(displayln (tuple-server-get c "user0"))

(displayln (tuple-server-get d "user2"))

(displayln (tuple-server-get d "user0"))

(displayln (tuple-server-get c "user2"))

)

(after-seconds 8

(node-send-exit remote-node))

(after-seconds 10

(exit 0))))

Figure 1: examples/named/master.rkt

spawn-remote-racket-node 原语连接到 "localhost" 并在那里启动一个 racloud 节点，该节点监听端口 6344 以接收进一步的指令。新 racloud 节点的句柄被 赋值给 remote-node 变量。使用 localhost 是为了 使示例可以在单台机器上运行。不过，localhost 可以替换为任何具有 ssh publickey 访问权限和 Racket 的主机。 supervise-place-at 在 remote-node 上创建一个新 place。 新 place 将在以后通过其名称 symbol 'tuple-server 来标识。 调用 dynamic-place 时，期望通过 tuple-path 模块路径和 'make-tuple-server symbol 返回一个 place 描述符。

tuple-server place 的代码存在于文件 "tuple.rkt" 中。"tuple.rkt" 文件包含了 define-named-remote-server 形式的使用，它定义了一个 适合由 supervise-place-at 调用的 RPC server。

#lang racket/base

(require racket/match

racket/place/define-remote-server)

(define-named-remote-server tuple-server

(define-state h (make-hash))

(define-rpc (set k v)

(hash-set! h k v)

v)

(define-rpc (get k)

(hash-ref h k #f))

(define-cast (hello)

(printf "Hello from define-cast\n")

(flush-output)))

Figure 2: examples/named/tuple.rkt

define-named-remote-server 形式接受一个标识符和一个 自定义表达式列表作为参数。通过在标识符前添加 make- 前缀来创建 place-thunk 函数。在本例中为 make-tuple-server。 make-tuple-server 标识符是传递给 上面 supervise-named-dynamic-place-at 形式的 place-function-name。define-state 自定义形式 被翻译为一个简单的 define 形式，该形式被 define-rpc 形式所封闭。

define-rpc 形式被展开为两个部分。第一部分 是调用 rpc 函数的客户端 stub。客户端 函数名通过连接 define-named-remote-server 标识符 tuple-server 和 RPC 函数名 set 形成 tuple-server-set。 RPC 客户端函数接受一个目标参数（即 remote-connection% 描述符），然后是 RPC 函数 参数。RPC 客户端函数通过调用内部函数 named-place-channel-put 将 RPC 函数名 set 和 RPC 参数发送到目标。然后 RPC 客户端 调用 named-place-channel-get 等待 RPC 响应。

define-rpc 的第二个展开部分是 RPC 调用的服务器端 实现。服务器通过 make-tuple-server 函数内部的 match 表达式来实现。tuple-server-set 的 match 子句匹配以 'set symbol 开头的消息。服务器使用 传递的参数执行 RPC 调用并将结果发送回 RPC 客户端。

define-cast 形式类似于 define-rpc 形式， 区别在于服务器不会向客户端发送回复消息

(module tuple racket/base

(require racket/place

racket/match)

(define/provide

(tuple-server-set dest k v)

(named-place-channel-put dest (list 'set k v))

(named-place-channel-get dest))

(define/provide

(tuple-server-get dest k)

(named-place-channel-put dest (list 'get k))

(named-place-channel-get dest))

(define/provide

(tuple-server-hello dest)

(named-place-channel-put dest (list 'hello)))

(define/provide

(make-tuple-server ch)

(let ()

(define h (make-hash))

(let loop ()

(define msg (place-channel-get ch))

(define (log-to-parent-real

msg

#:severity (severity 'info))

(place-channel-put

ch

(log-message severity msg)))

(syntax-parameterize

((log-to-parent (make-rename-transformer

#'log-to-parent-real)))

(match

msg

((list (list 'set k v) src)

(define result (let () (hash-set! h k v) v))

(place-channel-put src result)

(loop))

((list (list 'get k) src)

(define result (let () (hash-ref h k #f)))

(place-channel-put src result)

(loop))

((list (list 'hello) src)

(define result

(let ()

(printf "Hello from define-cast\n")

(flush-output)))

(loop))))

loop))))

Figure 3: define-named-remote-server 的展开