20. Process

概述

首先，关于线程的理解，如果有操作系统的相关知识，那么就很好理解。在操作系统中，线程是进程的一个子集。进程的定义：一个程序的一次运行。而线程则是，多个共享数据的进程。在本文中，线程表示的是在单个 Lisp 进程运行的多个单独的程序，其共享同一部分内存地址。一般来说，系统（不论是 lisp 内核 还是操作系统）会在这写线程间自动切换，因此这些线程看起来是并行运行的（异步）。本章要讲解的是线程的创建和管理，以及一些线程间的交互方式。想要了解更多关于 lisp 和其他进程之间的交互的话，参考阅读第 18 章：系统交互

对一些粗心的人要提醒下，大部分解释其会将线程称为进程——这是历史遗留问题，因为 Lisp 的出现要比术语线程这个词要早。这可以算作是成熟的解释器的标志之一。

ANSI Common Lisp 标准中没有涉及到这个话题。接下来会来讲解关于 bordeaux-threads、SBCL threads、lparallel 库。

比较常用的标准库是 —— 波多尔线程（bodeaux-threads），该库是由底层原语实现的。lparallel 就是在波多尔线程上构建的，同时实现了以下的特性：

• 简单的接收队列自任务模型
• 详细的并行结构
• 异步处理线程边界异常
• 并行版的 map、reduce、sort、remove 等
• promises, futures, and delayed evaluation constructs
• 用于并行化相互关联的任务的计算树
• 绑定与解绑 FIFO 队列
• 通道
• 任务优先级
• 按类别结束任务
• 集成超时

更多关于并行和并发的库，参见 awesome CL list
及 Quickdocs.

缘由

首先要问的是：为什么需要线程/进程呢？你可能会简单的回答说“我的引用很简单，根本不需要关注这些。”同在大部分情况下，很难想象没有多线程的复杂程序要怎么设计。例如：

• 需要开发个服务，这个服务（例如 websockets 章节中的 web 服务）要同时响应多个用户/连接
• 需要执行后台活动，同时不停止主程序。
• 需要在一段时间后通知应用
• 需要在等待系统资源时保持程序的运行活动
• 需要与多线程的系统（比如说，Windows 下的窗口通常都是在自己的线程中运行）进行通信
• 需要在应用的不同部分关联不同的上下文（如不同的动态绑定）
• 可能就是简单的像同时做两件事

并发与并行

鸣谢：以下部分首次是有 Timmy Jose 在 z0ltan.wordpress.com 上发布的。

instance) and context-switching them.
并发是同时运行不同（也许相关）的任务。也就是说，即便是在单核 CPU 的电脑上，也可以用线程的上下文切换来模拟同时性。

在系统（原生操作系统）中，线程的调度和上下文切换最终是由操作系统终止的。Java 和 Common Lisp 中也一样。

线程的调度完全是由程序来控制管理的线程叫“绿色（green）“线程。Erlang 就是个很好的例子。

那么并发和并行有什么差别呢？严格意义上来说，并行是同时在不同处理器或不同的核上平行地运行单个任务。狭义上来说，在单核单处理器的电脑上无法做到并行。

在更抽象的层面能更好的区分这两个概念：并发主要给客户端一种同时性的假象，因此在长操作中也不会出现系统锁定的现象。GUI 系统就是个很好的例子。因此并发关注的是用户体验，不关心性能。

Java Swing 工具套件和 JavaScript 都是单线程的，但它们都有同时性的表现，因为会在后台进行上下文切换。当然咯，大部分情况下并发也是由多线程/处理器实现的。

另一方面，并行，关注点是性能。例如，当要计算出一定范围内偶数的平方时，就可以一个大范围分成多个小块，然后再在不同的核或处理器上并行运行，再将结果给组合起来就是最终结果了。这就是 Map-Reduce 的操作。

现在已经在抽象上区分了并发和并行，之后就是实际上实现的机制。很多人在这对容易搞混。他们倾向于用具体的实现方法来约束抽象概念。从本质上讲，这两个抽象概念可以相同的机制实现！例如，可以使用Java 中相同的基本线程机制来实现并发特性和并行特性。只有抽象层面上，任务概念的相互作用或独立性才有影响。

例如，有个要使用不同的线程（可能是在不同的核/处理器上）才能完成部分工作的任务，但生成这个线程的线程逻辑上依赖衍生线程（不必阻塞该进程）的结果时，这还是并发。

所以最终的结论是：并发和并行是不同的概念，但可以使用相同的机制（如线程、进程等）实现。

波多尔线程

bordeaux 库可以不依赖平台进行基础的线程操作，同时也可以在不同的解释器上运行。有趣的是 bordeaux 是创建原生的线程，而是完全依赖解释器下层来创建线程。

另一方面，bordeaux 在底层线程的抽象中提供了一些很实用特性。

同时，在 demo 程序中，bordeaux 中的函数和 SBCL 中的函数很像。不知道这是不是个巧合。

在下文中会有详细的介绍（具体参见“封装”一节）

安装波多尔线程

通过 quicklisp 安装加载波多尔线程：

CL-USER> (ql:quickload :bt-semaphore)
To load "bt-semaphore":
  Load 1 ASDF system:
    bt-semaphore
; Loading "bt-semaphore"
(:BT-SEMAPHORE)

查看是否支持线程

不管使用的是哪个 Common Lisp 解释器，都可以使用标准的方法来检查是否支持线程：

CL-USER> (member :thread-support *FEATURES*)
(:THREAD-SUPPORT :SWANK :QUICKLISP :ASDF-PACKAGE-SYSTEM :ASDF3.1 :ASDF3 :ASDF2
 :ASDF :OS-MACOSX :OS-UNIX :NON-BASE-CHARS-EXIST-P :ASDF-UNICODE :64-BIT
 :64-BIT-REGISTERS :ALIEN-CALLBACKS :ANSI-CL :ASH-RIGHT-VOPS :BSD
 :C-STACK-IS-CONTROL-STACK :COMMON-LISP :COMPARE-AND-SWAP-VOPS
 :COMPLEX-FLOAT-VOPS :CYCLE-COUNTER :DARWIN :DARWIN9-OR-BETTER :FLOAT-EQL-VOPS
 :FP-AND-PC-STANDARD-SAVE :GENCGC :IEEE-FLOATING-POINT :INLINE-CONSTANTS
 :INODE64 :INTEGER-EQL-VOP :LINKAGE-TABLE :LITTLE-ENDIAN
 :MACH-EXCEPTION-HANDLER :MACH-O :MEMORY-BARRIER-VOPS :MULTIPLY-HIGH-VOPS
 :OS-PROVIDES-BLKSIZE-T :OS-PROVIDES-DLADDR :OS-PROVIDES-DLOPEN
 :OS-PROVIDES-PUTWC :OS-PROVIDES-SUSECONDS-T :PACKAGE-LOCAL-NICKNAMES
 :PRECISE-ARG-COUNT-ERROR :RAW-INSTANCE-INIT-VOPS :SB-DOC :SB-EVAL :SB-LDB
 :SB-PACKAGE-LOCKS :SB-SIMD-PACK :SB-SOURCE-LOCATIONS :SB-TEST :SB-THREAD
 :SB-UNICODE :SBCL :STACK-ALLOCATABLE-CLOSURES :STACK-ALLOCATABLE-FIXED-OBJECTS
 :STACK-ALLOCATABLE-LISTS :STACK-ALLOCATABLE-VECTORS
 :STACK-GROWS-DOWNWARD-NOT-UPWARD :SYMBOL-INFO-VOPS :UD2-BREAKPOINTS :UNIX
 :UNWIND-TO-FRAME-AND-CALL-VOP :X86-64)

如果解释器不支持线程的话，上面的代码会返回 “NIL”。

每个库都有自己的检查是否支持并发的方法，这就可可以不用上面的通用检查方法来检查了。

例如，在下面的例子中，使用的是 bordeaux 库。可以查看 supports-threads-p 这个全局变量来检查是否支持多线程，如果 support-threads-p 是 NIL，则不支持，T 的话就是支持：

CL-USER> bt:*supports-threads-p*
T

好了，现在已经知道这个方法了，那么现在来试试这个平台无关库（bordeaxu）和特定的平台（SBCL）吧。

为了这点，先看一些简单的例子吧：

• 基础 —— 列出当前线程、所有线程以及获取线程名
• 更新全局变量
• 在顶层打印消息
• 打印消息 —— 修复版
• 打印消息 —— 进阶版
• 修改多线程共享资源
• 修改共享资源 —— 有锁版
• 修改共享资源 —— 原子操作版
• 阻塞，销毁进程

基础 —— 列出当前线程、所有线程以及获取线程名

    ;;; Print the current thread, all the threads, and the current thread's name
    (defun print-thread-info ()
      (let* ((curr-thread (bt:current-thread))
             (curr-thread-name (bt:thread-name curr-thread))
             (all-threads (bt:all-threads)))
        (format t "Current thread: ~a~%~%" curr-thread)
        (format t "Current thread name: ~a~%~%" curr-thread-name)
        (format t "All threads:~% ~{~a~%~}~%" all-threads))
      nil)

输入如下：

    CL-USER> (print-thread-info)
    Current thread: #<THREAD "repl-thread" RUNNING {10043B8003}>
    Current thread name: repl-thread
    All threads:
     #<THREAD "repl-thread" RUNNING {10043B8003}>
    #<THREAD "auto-flush-thread" RUNNING {10043B7DA3}>
    #<THREAD "swank-indentation-cache-thread" waiting on: #<WAITQUEUE  {1003A28103}> {1003A201A3}>
    #<THREAD "reader-thread" RUNNING {1003A20063}>
    #<THREAD "control-thread" waiting on: #<WAITQUEUE  {1003A19E53}> {1003A18C83}>
    #<THREAD "Swank Sentinel" waiting on: #<WAITQUEUE  {1003790043}> {1003788023}>
    #<THREAD "main thread" RUNNING {1002991CE3}>
    NIL

从线程中更新全局变量：

    (defparameter *counter* 0)
    (defun test-update-global-variable ()
      (bt:make-thread
       (lambda ()
         (sleep 1)
         (incf *counter*)))
      *counter*)

可以使用 bt:make-thread 函数创建新新线程，该函数的参数是个抽象的 lambda 表达式。注意，该 lambda 表达式不能有参数。

还有一点需要注意的是，与其他语言不同的是，线程对象的创建和执行之间没有分隔。也就是说，线程创建好后就会直接执行。

输出如下：

    CL-USER> (test-update-global-variable)
    0
    CL-USER> *counter*
    1

如上所示，主线程立即返回了 *counter* 的初始值 0，大概在一秒钟后，匿名线程就会将该值跟新到 1。

线程创建：打印消息

    ;;; Print a message onto the top-level using a thread
    (defun print-message-top-level-wrong ()
      (bt:make-thread
       (lambda ()
         (format *standard-output* "Hello from thread!"))
       :name "hello")
      nil)

输出为：

    CL-USER> (print-message-top-level-wrong)
    NIL

哪里错了呢？问题出在变量绑定上。在这里，将 format 函数的参数设为 t 的话，输出就会指向的是顶层（Common Lisp 终端控制流中的常见术语），也指向了全局变量 *standard-output*。因此就可以在控制台中看到输出。

如果在同一个线程中运行上面的代码的话，是完全没有问题的。但当在不同的线程中，每个线程都会有单独的栈来保存变量。因此，即便是全局变量 *standard-output*（假设所有的线程中都有）也会在每个线程中重新绑定！这和 Java 中的 ThreadLocal 存储概念很相似。现在需要对上面的代码进行修复。

但是要怎么来修复这个问题呢？当然是在线程创建时将输出绑定到顶层咯。纯词法作用域是可以挽救以下这种情况的！

    ;;; Print a message onto the top-level using a thread — fixed
    (defun print-message-top-level-fixed ()
      (let ((top-level *standard-output*))
        (bt:make-thread
         (lambda ()
           (format top-level "Hello from thread!"))
         :name "hello")))
      nil)

输出结果是：

    CL-USER> (print-message-top-level-fixed)
    Hello from thread!
    NIL

哎唷，可以了。但是，下面还有个好玩的宏也能返回同样的结果。

打印消息 —— read-time eval macro

先看代码：

    ;;; Print a message onto the top-level using a thread - reader macro
    (eval-when (:compile-toplevel)
      (defun print-message-top-level-reader-macro ()
        (bt:make-thread
         (lambda ()
           (format #.*standard-output* "Hello from thread!")))
        nil))
    (print-message-top-level-reader-macro)

输出结果是：

    CL-USER> (print-message-top-level-reader-macro)
    Hello from thread!
    NIL

结果和上面是一样的，但 *standard-output* 前面奇怪的 #. 是做什么的呢？

eval-when 控制什么时候对 Lisp 表达式求值。这里有三个 target：:compile-toplevel、:load-toplevel 和 :execute

#. 会调用 “Reader 宏”。因为这个宏在 Common Lisp 中有特殊含义的，所以被叫做 reader（或 read）宏。其特殊性在于负责会读取 Common Lisp 表达式然后解析读入的表达式。这个特殊的 reader 宏确保在读取时绑定 *standard-output*。

在读取时绑定值能够确保线程在运行时绑定的是原始值 *standard-output*，这样才能在顶层输出。

现在，是 eval-when 发挥作用的时候了。将整个函数定义封装到 eval-when 中，确保编译时间会进行求值替换，*standard-output* 也绑定到正确的值。但如果不用 eval-when 的话，就会看到下面的错误：

      error:
        don't know how to dump #<SWANK/GRAY::SLIME-OUTPUT-STREAM {100439EEA3}> (default MAKE-LOAD-FORM method called).
        ==>
          #<SWANK/GRAY::SLIME-OUTPUT-STREAM {100439EEA3}>
      note: The first argument never returns a value.
      note:
        deleting unreachable code
        ==>
          "Hello from thread!"
    Compilation failed.

这就解释的通了，由于输出流返回的结果是个流而不是个定义好的值（format 函数所期望的），所以 SBCL 无法解析这个结果。这就是“unreachable code” 错误的原因。

注意，如果直接在 REPL 中直接运行代码的话是完全没问题的，因为 REPL 线程会正确的处理所有的符号。

修改共享资源

假设有个简单的 bank-account 类（无错误检查）：

    ;;; Modify a shared resource from multiple threads
    (defclass bank-account ()
      ((id :initarg :id
           :initform (error "id required")
           :accessor :id)
       (name :initarg :name
             :initform (error "name required")
             :accessor :name)
       (balance :initarg :balance
                :initform 0
                :accessor :balance)))
    (defgeneric deposit (account amount)
      (:documentation "Deposit money into the account"))
    (defgeneric withdraw (account amount)
      (:documentation "Withdraw amount from account"))
    (defmethod deposit ((account bank-account) (amount real))
      (incf (:balance account) amount))
    (defmethod withdraw ((account bank-account) (amount real))
      (decf (:balance account) amount))

同时有个客户端，该客户端不信任任何形式的同步：

    (defparameter *rich*
      (make-instance 'bank-account
                     :id 1
                     :name "Rich"
                     :balance 0))
    ; compiling (DEFPARAMETER *RICH* ...)
    (defun demo-race-condition ()
      (loop repeat 100
         do
           (bt:make-thread
            (lambda ()
              (loop repeat 10000 do (deposit *rich* 100))
              (loop repeat 10000 do (withdraw *rich* 100))))))

来看下下面这些操作：创建个新的银行账户实例（账户余额（balance）为 0），然后创建 100 个线程，每个线程会执行 10000 次存入 100 额度的操作，之后在执行相同次数的提取操作。最终的结果应该是和开户时一样，都是 0，对吧？那么现在来检查下对不对。

将示例运行后，可能会得到如下的结果：

    CL-USER> (:balance *rich*)
    0
    CL-USER> (dotimes (i 5)
               (demo-race-condition))
    NIL
    CL-USER> (:balance *rich*)
    22844600

哇！造成这个差异的原因是 incf 和 decf 不是原子操作——它们是分为好几个步骤操作的，而且执行的顺序也不受控制。

这种现象叫做“竞争条件” —— 多个线程同时竞争相同的共享资源，一个值在修改的同时被读取，这得到的就是个错误的值。那要怎么修复这个问题呢？简单的办法就是使用锁（本例中使用的是互斥锁（mutex），在复杂的情况下使用信号量（semaphores））

修改共享资源：修复版（使用锁）

现将之前的账户余额该会到 0：

    CL-USER> (setf (:balance *rich*) 0)
    0
    CL-USER> (:balance *rich*)
    0

然后在 demo-race-condition 函数中对共享资源加上锁（bt:make-lock 创建锁）：

    (defvar *lock* (bt:make-lock))
    ; compiling (DEFVAR *LOCK* …)
    (defun demo-race-condition-locks ()
      (loop repeat 100
         do
           (bt:make-thread
            (lambda ()
              (loop repeat 10000 do (bt:with-lock-held (*lock*)
                                      (deposit *rich* 100)))
              (loop repeat 10000 do (bt:with-lock-held (*lock*)
                                      (withdraw *rich* 100)))))))
    ; compiling (DEFUN DEMO-RACE-CONDITION-LOCKS ...)

然后执行和上面相同的操作：

    CL-USER> (dotimes (i 100)
               (demo-race-condition-locks))
    NIL
    CL-USER> (:balance *rich*)
    0

很好！现在这个就要好多了。当然，记住互斥锁的使用影响性能。在这种情况下还有个更好的方法 —— 尽可能的使用原子操作。下面就来讲解吧。

修改共享资源：原子操作版

原子操作是操作系统为保证内部事物执行的操作，也就是说，主任务中的子任务不会受外部影响。操作要么成功，要么失败，没有中间层，也没有不一致的状态。

还有一点是，在保护共享资源的访问方面，原子操作的性能要远远优于比互斥锁。这个在 demo 中可以看到差别。

bordeaux 库并不支持原子操作，因此需要依赖于解释器来实现这个。在本章的例子中，使用的是 SBCL，在下面的 SBCL 章节中会有相应的 demo。

线程阻塞、销毁

阻塞线程使用 bt:join-thread 函数，销毁线程（不推荐）的话使用 bt:destory-thread 函数。

一个简单的 demo：

    (defmacro until (condition &body body)
      (let ((block-name (gensym)))
        `(block ,block-name
           (loop
               (if ,condition
                   (return-from ,block-name nil)
                   (progn
                       ,@body))))))
    (defun join-destroy-thread ()
      (let* ((s *standard-output*)
            (joiner-thread (bt:make-thread
                            (lambda ()
                              (loop for i from 1 to 10
                                 do
                                   (format s "~%[Joiner Thread]  Working...")
                                   (sleep (* 0.01 (random 100)))))))
            (destroyer-thread (bt:make-thread
                               (lambda ()
                                 (loop for i from 1 to 1000000
                                    do
                                      (format s "~%[Destroyer Thread] Working...")
                                      (sleep (* 0.01 (random 10000))))))))
        (format t "~%[Main Thread] Waiting on joiner thread...")
        (bt:join-thread joiner-thread)
        (format t "~%[Main Thread] Done waiting on joiner thread")
        (if (bt:thread-alive-p destroyer-thread)
            (progn
              (format t "~%[Main Thread] Destroyer thread alive... killing it")
              (bt:destroy-thread destroyer-thread))
            (format t "~%[Main Thread] Destroyer thread is already dead"))
        (until (bt:thread-alive-p destroyer-thread)
               (format t "[Main Thread] Waiting for destroyer thread to die..."))
        (format t "~%[Main Thread] Destroyer thread dead")
        (format t "~%[Main Thread] Adios!~%")))

运行后的结果如下：

    CL-USER> (join-destroy-thread)
    [Joiner Thread]  Working...
    [Destroyer Thread] Working...
    [Main Thread] Waiting on joiner thread...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Main Thread] Done waiting on joiner thread
    [Main Thread] Destroyer thread alive... killing it
    [Main Thread] Destroyer thread dead
    [Main Thread] Adios!
    NIL

只有当条件为真时，until 才会退出循环。剩下的代码就不用详细介绍了——主线程等待阻塞线程完成，然后立即销毁。

再次声明，不推荐使用 bt:destory-thread。任意要用 bt:destory-thread 函数的场景都可以找到一种更好的方法进行替代。

接下来，来看一些更完整的例子吧，这些例子会将目前所提到的概念全都联系在一起。

实用函数

下面是 demo 中用到的函数、宏和全局变量以及一些额外的内容的摘要，基本上涵盖了大多数编程场景：

• bt:*supports-thread-p* （检查是否支持基础线程操作）
• bt:make-thread （创建线程）
• bt:current-thread （返回当前线程对象）
• bt:all-threads （返回执行中的所有线程列表）
• bt:thread-alive-p （检查线程是否存活）
• bt:thread-name （返回线程名）
• bt:join-thread （阻塞线程）
• bt:interrupt-thread （中断线程）
• bt:destroy-thread （退出/销毁线程）
• bt:make-lock （创建互斥锁）
• bt:with-lock-held （线程锁）

SBCL 线程

SBCL 中的 sb-thread 包中支持原生多线程。里面都是低级函数，但可以在顶层构建自己的抽象，就像 demo 中展示的一样。

更多的细节可以查看下文的“封装”部分。

通过下面的例子可以看出，bordeaux 和 SBCL 关于线程的函数有很强的对应关系。在大部分情况中，只是将包的名字由 bt 改成 sb-thread。

很明显，Bordeaux 线程库或多或少是基于 SBCL 实现的。因此，只有在语法或语义上有很大差别时才会进行讲解。

基础 —— 列出当前线程、所有线程以及获取线程名

代码如下：

    ;;; Print the current thread, all the threads, and the current thread's name
    (defun print-thread-info ()
      (let* ((curr-thread sb-thread:*current-thread*)
             (curr-thread-name (sb-thread:thread-name curr-thread))
             (all-threads (sb-thread:list-all-threads)))
        (format t "Current thread: ~a~%~%" curr-thread)
        (format t "Current thread name: ~a~%~%" curr-thread-name)
        (format t "All threads:~% ~{~a~%~}~%" all-threads))
      nil)

输出结果是：

    CL-USER> (print-thread-info)
    Current thread: #<THREAD "repl-thread" RUNNING {10043B8003}>
    Current thread name: repl-thread
    All threads:
     #<THREAD "repl-thread" RUNNING {10043B8003}>
    #<THREAD "auto-flush-thread" RUNNING {10043B7DA3}>
    #<THREAD "swank-indentation-cache-thread" waiting on: #<WAITQUEUE  {1003A28103}> {1003A201A3}>
    #<THREAD "reader-thread" RUNNING {1003A20063}>
    #<THREAD "control-thread" waiting on: #<WAITQUEUE  {1003A19E53}> {1003A18C83}>
    #<THREAD "Swank Sentinel" waiting on: #<WAITQUEUE  {1003790043}> {1003788023}>
    #<THREAD "main thread" RUNNING {1002991CE3}>
    NIL

更新全局变量

代码如下：

    ;;; Update a global variable from a thread
    (defparameter *counter* 0)
    (defun test-update-global-variable ()
      (sb-thread:make-thread
       (lambda ()
         (sleep 1)
         (incf *counter*)))
      *counter*)

结果如下：

    CL-USER> (test-update-global-variable)
    0

打印消息

代码如下：

    ;;; Print a message onto the top-level using a thread
    (defun print-message-top-level-wrong ()
      (sb-thread:make-thread
       (lambda ()
         (format *standard-output* "Hello from thread!")))
      nil)

输出如下：

    CL-USER> (print-message-top-level-wrong)
    NIL

修复版：

    ;;; Print a message onto the top-level using a thread - fixed
    (defun print-message-top-level-fixed ()
      (let ((top-level *standard-output*))
        (sb-thread:make-thread
         (lambda ()
           (format top-level "Hello from thread!"))))
      nil)

结果如下：

    CL-USER> (print-message-top-level-fixed)
    Hello from thread!
    NIL

打印消息：优化版

代码：

    ;;; Print a message onto the top-level using a thread - reader macro
    (eval-when (:compile-toplevel)
      (defun print-message-top-level-reader-macro ()
        (sb-thread:make-thread
         (lambda ()
           (format #.*standard-output* "Hello from thread!")))
        nil))

结果：

    CL-USER> (print-message-top-level-reader-macro)
    Hello from thread!
    NIL

修改共享资源

代码：

    ;;; Modify a shared resource from multiple threads
    (defclass bank-account ()
      ((id :initarg :id
           :initform (error "id required")
           :accessor :id)
       (name :initarg :name
             :initform (error "name required")
             :accessor :name)
       (balance :initarg :balance
                :initform 0
                :accessor :balance)))
    (defgeneric deposit (account amount)
      (:documentation "Deposit money into the account"))
    (defgeneric withdraw (account amount)
      (:documentation "Withdraw amount from account"))
    (defmethod deposit ((account bank-account) (amount real))
      (incf (:balance account) amount))
    (defmethod withdraw ((account bank-account) (amount real))
      (decf (:balance account) amount))
    (defparameter *rich*
      (make-instance 'bank-account
                     :id 1
                     :name "Rich"
                     :balance 0))
    (defun demo-race-condition ()
      (loop repeat 100
         do
           (sb-thread:make-thread
            (lambda ()
              (loop repeat 10000 do (deposit *rich* 100))
              (loop repeat 10000 do (withdraw *rich* 100))))))

结果：

    CL-USER> (:balance *rich*)
    0
    CL-USER> (demo-race-condition)
    NIL
    CL-USER> (:balance *rich*)
    3987400

修改共享资源：修复版（互斥锁）

代码：

    (defvar *lock* (sb-thread:make-mutex))
    (defun demo-race-condition-locks ()
      (loop repeat 100
         do
           (sb-thread:make-thread
            (lambda ()
              (loop repeat 10000 do (sb-thread:with-mutex (*lock*)
                                      (deposit *rich* 100)))
              (loop repeat 10000 do (sb-thread:with-mutex (*lock*)
                                      (withdraw *rich* 100)))))))

唯一不同的是在 bordeaux 中用的是 make-lock，这里用的是 make-mutex，配合示例中的宏 with-mutex 使用。

输出结果是：

    CL-USER> (:balance *rich*)
    0
    CL-USER> (demo-race-condition-locks)
    NIL
    CL-USER> (:balance *rich*)
    0

修改共享资源：原子操作版

首先，代码如下：

    ;;; Modify a shared resource from multiple threads - atomics
    (defgeneric atomic-deposit (account amount)
      (:documentation "Atomic version of the deposit method"))
    (defgeneric atomic-withdraw (account amount)
      (:documentation "Atomic version of the withdraw method"))
    (defmethod atomic-deposit ((account bank-account) (amount real))
      (sb-ext:atomic-incf (car (cons (:balance account) nil)) amount))
    (defmethod atomic-withdraw ((account bank-account) (amount real))
      (sb-ext:atomic-decf (car (cons (:balance account) nil)) amount))
    (defun demo-race-condition-atomics ()
      (loop repeat 100
         do (sb-thread:make-thread
             (lambda ()
               (loop repeat 10000 do (atomic-deposit *rich* 100))
               (loop repeat 10000 do (atomic-withdraw *rich* 100))))))

结果是：

    CL-USER> (dotimes (i 5)
               (format t "~%Opening: ~d" (:balance *rich*))
               (demo-race-condition-atomics)
               (format t "~%Closing: ~d~%" (:balance *rich*)))
    Opening: 0
    Closing: 0
    Opening: 0
    Closing: 0
    Opening: 0
    Closing: 0
    Opening: 0
    Closing: 0
    Opening: 0
    Closing: 0
    NIL

如你所见，SBCL 的原子操作优点怪。这里使用到了两个函数：sb-ext:incf 和 sb-ext:atomic-decf，它们的格式分别为：

Macro: atomic-incf [sb-ext] place &optional diff

和

Macro: atomic-decf [sb-ext] place &optional diff

比较有趣的地方是 “place” 参数需要是以下中一个（根据文件）：

• 具有声明类型（64位无符号字节类型）的 defstruct 属性或 simple-array（64为无符号字节类型）的 aref。 sb-ext:word 类型可用于这些目的。
• cons 中的 car 或 cdr 类型（对应第一个和剩下的）
• defglobal 定义的已声明的修正数（fixnum）变量

这就是为什么会在 atomic-deposit 和 atomic-decf 方法中使用 bizarre 结构。

尽可能的使用原子操作是考虑到性能方面。现在快速地将 demo-race-condition-locks 和 demo-race-condition-atomics 函数分别执行 1000 次，然后查看它们的性能上的差别（如果有的话）：

使用互斥锁：

    CL-USER> (time
                        (loop repeat 100
                          do (demo-race-condition-locks)))
    Evaluation took:
      57.711 seconds of real time
      431.451639 seconds of total run time (408.014746 user, 23.436893 system)
      747.61% CPU
      126,674,011,941 processor cycles
      3,329,504 bytes consed
    NIL

使用原子操作：

    CL-USER> (time
                        (loop repeat 100
                         do (demo-race-condition-atomics)))
    Evaluation took:
      2.495 seconds of real time
      8.175454 seconds of total run time (6.124259 user, 2.051195 system)
      [ Run times consist of 0.420 seconds GC time, and 7.756 seconds non-GC time. ]
      327.66% CPU
      5,477,039,706 processor cycles
      3,201,582,368 bytes consed
    NIL

那么结果呢？互斥锁版本大概用了 57s，而原子操作只用的 2s！ 这个简直是天差地别！

线程阻塞、销毁

代码如下：

    ;;; Joining on and destroying a thread
    (defmacro until (condition &body body)
      (let ((block-name (gensym)))
        `(block ,block-name
           (loop
               (if ,condition
                   (return-from ,block-name nil)
                   (progn
                       ,@body))))))
    (defun join-destroy-thread ()
      (let* ((s *standard-output*)
            (joiner-thread (sb-thread:make-thread
                            (lambda ()
                              (loop for i from 1 to 10
                                 do
                                   (format s "~%[Joiner Thread]  Working...")
                                   (sleep (* 0.01 (random 100)))))))
            (destroyer-thread (sb-thread:make-thread
                               (lambda ()
                                 (loop for i from 1 to 1000000
                                    do
                                      (format s "~%[Destroyer Thread] Working...")
                                      (sleep (* 0.01 (random 10000))))))))
        (format t "~%[Main Thread] Waiting on joiner thread...")
        (bt:join-thread joiner-thread)
        (format t "~%[Main Thread] Done waiting on joiner thread")
        (if (sb-thread:thread-alive-p destroyer-thread)
            (progn
              (format t "~%[Main Thread] Destroyer thread alive... killing it")
              (sb-thread:terminate-thread destroyer-thread))
            (format t "~%[Main Thread] Destroyer thread is already dead"))
        (until (sb-thread:thread-alive-p destroyer-thread)
               (format t "[Main Thread] Waiting for destroyer thread to die..."))
        (format t "~%[Main Thread] Destroyer thread dead")
        (format t "~%[Main Thread] Adios!~%")))

结果为：

    CL-USER> (join-destroy-thread)
    [Joiner Thread]  Working...
    [Destroyer Thread] Working...
    [Main Thread] Waiting on joiner thread...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Joiner Thread]  Working...
    [Main Thread] Done waiting on joiner thread
    [Main Thread] Destroyer thread alive... killing it
    [Main Thread] Destroyer thread dead
    [Main Thread] Adios!
    NIL

实用函数

下面是示例中用到的函数、宏和全局变量以及一些额外的内容的摘要，基本上涵盖了大多数编程场景：

• (member :thread-support *features*) (检查是否支持多线程）
• sb-thread:make-thread （线程创建）
• sb-thread:*current-thread* （获取当前线程对象）
• sb-thread:list-all-threads （返回所有运行中的线程）
• sb-thread:thread-alive-p （检查线程是否存活）
• sb-thread:thread-name （返回线程名）
• sb-thread:join-thread （线程阻塞）
• sb-thread:interrupt-thread （线程中断）
• sb-thread:destroy-thread （退出/销毁线程）
• sb-thread:make-mutex （创建互斥锁）
• sb-thread:with-mutex （线程锁）

封装

如你所见，Common Lisp 对并行的支持是很简单的，这主要是因为 ANSI Common Lisp 规范中缺少这个重要的特性。但这并没有减少 Common Lisp 实现，也没有减少像 bordeaux 这样的库。

可以自行选择深入阅读相关内容。以下是我自己的参考：

• Common Lisp Recipes
• Bordeaux API Reference
• SBCL Manual
• The Common Lisp Hyperspec

接下来，是关于这个迷你系列的最后一片文章：在 Common Lisp 中使用 lparallel 库。

并行编程：lparallel

值得注意的是，lparallel 也支持异步编程，而不是个纯并行库。像之前所说的那样，并行只是个抽象的概念，在并行中，任务之间是相互独立的。

lparallel 库基于 bordeaux 线程库构建的。

像之前提到的那样，并行和并发可以通过相同的方法实现（一般就是这样的）：线程、进程等。唯一不同的就是概念上的差别。

注意本文中的例子不一定都是并行的。通常来说，像 Promises 和 Futures 中的异步构造更适合并发。

lparallel 库的操作方式如下（基本情况）：

• 通过 lparallel:make-kernel 调用内核来创建示例。其中内核就是调度和执行任务的组件。
• 按照 futures、promises 和其他高级函数概念设计代码。为了实现这些，lparallel 支持 channels、promises、future 和 cognate。
• 使用 lparallel 库调用 cognates 执行操作，其中 cognates 在 Common Lisp 中有等价的函数。比如说，并行函数 lparallel:pmap 等价于 Common Lisp 中的 map。
• 最后，关掉第一步中通过 kernel 创建的实例 lparallel:end-kernel。

注意，确保运行中的任务需要符合逻辑，以及处理所有可变状态是由开发人员负责。

鸣谢：本文首次出现在 z0ltan.wordpress.com.

安装

先来看看能不能用 Quicklisp 安装 lparallel：

CL-USER> (ql:system-apropos "lparallel")
#<SYSTEM lparallel / lparallel-20160825-git / quicklisp 2016-08-25>
#<SYSTEM lparallel-bench / lparallel-20160825-git / quicklisp 2016-08-25>
#<SYSTEM lparallel-test / lparallel-20160825-git / quicklisp 2016-08-25>
; No value

看起来是可以，那么直接安装吧：

CL-USER> (ql:quickload :lparallel)
To load "lparallel":
  Load 2 ASDF systems:
    alexandria bordeaux-threads
  Install 1 Quicklisp release:
    lparallel
; Fetching #<URL "http://beta.quicklisp.org/archive/lparallel/2016-08-25/lparallel-20160825-git.tgz">
; 76.71KB
==================================================
78,551 bytes in 0.62 seconds (124.33KB/sec)
; Loading "lparallel"
[package lparallel.util]..........................
[package lparallel.thread-util]...................
[package lparallel.raw-queue].....................
[package lparallel.cons-queue]....................
[package lparallel.vector-queue]..................
[package lparallel.queue].........................
[package lparallel.counter].......................
[package lparallel.spin-queue]....................
[package lparallel.kernel]........................
[package lparallel.kernel-util]...................
[package lparallel.promise].......................
[package lparallel.ptree].........................
[package lparallel.slet]..........................
[package lparallel.defpun]........................
[package lparallel.cognate].......................
[package lparallel]
(:LPARALLEL)

下面就来看看怎么使用 lparallel 库吧。

序：调用 CFFI 获取 CPU 核的数量

首先，要先了解要在并行的例子中使用的线程数。理想状况下，线程与 CPU 核的最好为 1：1。

cffi 库可以实现这个目标。本来是准备为 cffi 库写篇详细的文章，但现在，还是继续本章的话题吧：

安装 CFFI:

CL-USER> (ql:quickload :cffi)
To load "cffi":
  Load 4 ASDF systems:
    alexandria babel trivial-features uiop
  Install 1 Quicklisp release:
    cffi
; Fetching #<URL "http://beta.quicklisp.org/archive/cffi/2016-03-18/cffi_0.17.1.tgz">
; 234.48KB
==================================================
240,107 bytes in 5.98 seconds (39.22KB/sec)
; Loading "cffi"
[package cffi-sys]................................
[package cffi]....................................
..................................................
[package cffi-features]
(:CFFI)

用 C 语言获取计算机上的逻辑核数量：

#include <stdio.h>
#include <sys/types.h>
#include <sys/sysctl.h>
int get_core_count();
int main()
{
    printf("%d\n", get_core_count());
    return 0;
}
int32_t get_core_count()
{
    const char* s = "hw.logicalcpu";
    int32_t core_count;
    size_t len = sizeof(core_count);
    sysctlbyname(s, &core_count, &len, NULL, 0);
    return core_count;
}

将上面 C 代码编译成共享库（注：本人使用的是 Mac OS X，使用的是 Clang。用 gcc 的话，请参考相关文档）：

Timmys-MacBook-Pro:Parallelism z0ltan$ clang -dynamiclib get_core_count.c -o libcorecount.dylib

在 Common Lisp 中调用这个函数：

CL-USER> (cffi:use-foreign-library "libcorecount.dylib")
#<CFFI:FOREIGN-LIBRARY LIBCORECOUNT.DYLIB-853 "libcorecount.dylib">
CL-USER> (cffi:foreign-funcall "get_core_count" :int)
8

可以看到结果计算机 CPU 的核（core）的数量是 8（这是正确的），可以通过下面的命令验证：

Timmys-MacBook-Pro:Parallelism z0ltan$ sysctl -n "hw.logicalcpu"

通用配置

在本例中，会讲解初始化配置，同时会介绍这些配置的作用。

加载库：

CL-USER> (ql:quickload :lparallel)
To load "lparallel":
  Load 1 ASDF system:
    lparallel
; Loading "lparallel"
(:LPARALLEL)

初始化 lparallel 内核：

CL-USER> (setf lparallel:*kernel* (lparallel:make-kernel 8 :name "custom-kernel"))
#<LPARALLEL.KERNEL:KERNEL :NAME "custom-kernel" :WORKER-COUNT 8 :USE-CALLER NIL :ALIVE T :SPIN-COUNT 2000 {1003141F03}>

注意：全局变量 *kernel* 可以重新赋值，这样就可以在同一运行期间共存多个内核。之后，定义 kernel 的相关信息：

CL-USER> (defun show-kernel-info ()
           (let ((name (lparallel:kernel-name))
                 (count (lparallel:kernel-worker-count))
                 (context (lparallel:kernel-context))
                 (bindings (lparallel:kernel-bindings)))
             (format t "Kernel name = ~a~%" name)
             (format t "Worker threads count = ~d~%" count)
             (format t "Kernel context = ~a~%" context)
             (format t "Kernel bindings = ~a~%" bindings)))
WARNING: redefining COMMON-LISP-USER::SHOW-KERNEL-INFO in DEFUN
SHOW-KERNEL-INFO
CL-USER> (show-kernel-info)
Kernel name = custom-kernel
Worker threads count = 8
Kernel context = #<FUNCTION FUNCALL>
Kernel bindings = ((*STANDARD-OUTPUT* . #<SLIME-OUTPUT-STREAM {10044EEEA3}>)
                   (*ERROR-OUTPUT* . #<SLIME-OUTPUT-STREAM {10044EEEA3}>))
NIL

终止内核（这点很重要，因为不手动结束内核的话， *kernel* 不会被垃圾回收掉）：

CL-USER> (lparallel:end-kernel :wait t)
(#<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {100723FA83}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {100723FE23}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {10072581E3}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {1007258583}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {1007258923}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {1007258CC3}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {1007259063}>
 #<SB-THREAD:THREAD "custom--kernel" FINISHED values: NIL {1007259403}>)

继续来看 lparallel 库的其他方面。

对于之后的 demo，从代码的角度会使用下面的初始设置：

(require ‘lparallel)
(require ‘bt-semaphore)
(defpackage :lparallel-user
  (:use :cl :lparallel :lparallel.queue :bt-semaphore))
(in-package :lparallel-user)
;;; initialise the kernel
(defun init ()
  (setf *kernel* (make-kernel 8 :name "channel-queue-kernel")))
(init)

所以内核使用的是 8 线程（可根据自己计算机自行设定）。

一旦运行完所有的例子后，需要运行下面代码来关闭内核并释放系统资源：

;;; shut the kernel down
(defun shutdown ()
  (end-kernel :wait t))
(shutdown)

通道（channels）和队列（queues）

首先，某些定义是有顺序的。

提交个内核的任务叫做 task。这只是个带参函数对象。

lparallel 中的通道（channel）和 Go 中通道的概念相似。只是与线程通信的一种方式。在本章的例子中，channel 就是个将任务提交给内核的特殊方法。

在 lparallel 中用 lparallel:make-channel 创建 channel。使用 lparallel:submit-task 提交任务，接受任务使用 lparallel:receive-result。

例如说，计算数字的平方值：

(defun calculate-square (n)
  (let* ((channel (lparallel:make-channel))
         (res nil))
    (lparallel:submit-task channel #'(lambda (x)
                                       (* x x))
                           n)
    (setf res (lparallel:receive-result channel))
    (format t "Square of ~d = ~d~%" n res)))

输出为：

LPARALLEL-USER> (calculate-square 100)
Square of 100 = 10000
NIL

现在，要将多个任务提交到统一个通道中。在下面的这个简单的例子中，只创建三个简单的任务，分别是计算输入的平方、三次放和四次方：

注意在多个任务重，输出结果的顺序并不确定：

(defun test-basic-channel-multiple-tasks ()
  (let ((channel (make-channel))
        (res '()))
    (submit-task channel #'(lambda (x)
                             (* x x))
                 10)
    (submit-task channel #'(lambda (y)
                             (* y y y))
                 10)
    (submit-task channel #'(lambda (z)
                             (* z z z z))
                 10)
     (dotimes (i 3 res)
       (push (receive-result channel) res))))

输出如下：

LPARALLEL-USER> (dotimes (i 3)
                              (print (test-basic-channel-multiple-tasks)))
(100 1000 10000)
(100 1000 10000)
(10000 1000 100)
NIL

为了将消息在运行中的线程中传递，lparallel 也支持创建阻塞队列。队列的创建使用的是 lparallel.queue:make-queue。

实用的队列函数：

• lparallel.queue:make-queue: create a FIFO blocking queue
• lparallel.queue:push-queue: insert an element into the queue
• lparallel.queue:pop-queue: pop an item from the queue
• lparallel.queue:peek-queue: inspect value without popping it
• lparallel.queue:queue-count: the number of entries in the queue
• lparallel.queue:queue-full-p: check if the queue is full
• lparallel.queue:queue-empty-p:check if the queue is empty
• lparallel.queue:with-locked-queue: lock the queue during access

下面是展示基础队列的属性的 demo：

    (defun test-queue-properties ()
      (let ((queue (make-queue :fixed-capacity 5)))
        (loop
           when (queue-full-p queue)
           do (return)
           do (push-queue (random 100) queue))
         (print (queue-full-p queue))
        (loop
           when (queue-empty-p queue)
           do (return)
           do (print (pop-queue queue)))
        (print (queue-empty-p queue)))
      nil)

执行后的结果为：

    LPARALLEL-USER> (test-queue-properties)
    T
    17
    51
    55
    42
    82
    T
    NIL

注：lparallel.queue:make-queue 是个通用接口，其后面实际上有多种类型队列。比如说，在上个例子中，基于 :fixed-capacify 关键词参数，真正的队列类型是 lparallel.vector-queue。

文档中没有指明要给 lparallel.queue:make-queue 传递什么参数，因此可以用下面的方法来确定：

    LPARALLEL-USER> (describe 'lparallel.queue:make-queue)
    LPARALLEL.QUEUE:MAKE-QUEUE
      [symbol]
    MAKE-QUEUE names a compiled function:
      Lambda-list: (&REST ARGS)
      Derived type: FUNCTION
      Documentation:
        Create a queue.
        The queue contents may be initialized with the keyword argument
        `initial-contents'.
        By default there is no limit on the queue capacity. Passing a
        `fixed-capacity' keyword argument limits the capacity to the value
        passed. `push-queue' will block for a full fixed-capacity queue.
      Source file: /Users/z0ltan/quicklisp/dists/quicklisp/software/lparallel-20160825-git/src/queue.lisp
    MAKE-QUEUE has a compiler-macro:
      Source file: /Users/z0ltan/quicklisp/dists/quicklisp/software/lparallel-20160825-git/src/queue.lisp
    ; No value

如上所示： lparallel.queue:make-queue 支持两个关键词参数：:fixed-capacity 和 initial-contents。

现在，如果给定参数为 :fixed-capacity，那么队列的实际类型是 lparallel.vector-queue。如果没有给定这个关键词参数，队列的默认类型是 lparallel.cons-queue（大小不限的队列），如下面代码所展示的那样：

    (defun check-queue-types ()
      (let ((queue-one (make-queue :fixed-capacity 5))
            (queue-two (make-queue)))
        (format t "queue-one is of type: ~a~%" (type-of queue-one))
        (format t "queue-two is of type: ~a~%" (type-of queue-two))))
    LPARALLEL-USER> (check-queue-types)
    queue-one is of type: VECTOR-QUEUE
    queue-two is of type: CONS-QUEUE
    NIL

当然，你总是能指定创建示例队列的类型，但如果可能的话，最好还是坚持使用通用接口然后让库来创建适合的类型。

现在，来看看下面的代码吧：

    (defun test-basic-queue ()
      (let ((queue (make-queue))
            (channel (make-channel))
            (res '()))
        (submit-task channel #'(lambda ()
                         (loop for entry = (pop-queue queue)
                            when (queue-empty-p queue)
                            do (return)
                            do (push (* entry entry) res))))
        (dotimes (i 100)
          (push-queue i queue))
        (receive-result channel)
        (format t "~{~d ~}~%" res)))

在上面代码中，提交单个任务，该任务重复扫描队列，直到队列为空，提出（pop）其中的值，让后把提出的值雅茹 res 列表。

输出为：

    LPARALLEL-USER> (test-basic-queue)
    9604 9409 9216 9025 8836 8649 8464 8281 8100 7921 7744 7569 7396 7225 7056 6889 6724 6561 6400 6241 6084 5929 5776 5625 5476 5329 5184 5041 4900 4761 4624 4489 4356 4225 4096 3969 3844 3721 3600 3481 3364 3249 3136 3025 2916 2809 2704 2601 2500 2401 2304 2209 2116 2025 1936 1849 1764 1681 1600 1521 1444 1369 1296 1225 1156 1089 1024 961 900 841 784 729 676 625 576 529 484 441 400 361 324 289 256 225 196 169 144 121 100 81 64 49 36 25 16 9 4 1 0
    NIL

终止任务

这是就适合使用 lparallel:kill-task 函数。这个函数在任务无响应时使用。lparallel 文档明确的指出杀死任务是最终手段。

创建的任务都会分配到 :default 中。动态属性 *task-category 用来存储这个值，且 task-category* 也可用动态地绑定到不同的值（如下所示）：

;;; kill default tasks
(defun test-kill-all-tasks ()
  (let ((channel (make-channel))
        (stream *query-io*))
    (dotimes (i 10)
      (submit-task channel #'(lambda (x)
                               (sleep (random 10))
                               (format stream "~d~%" (* x x))) (random 10)))
    (sleep (random 2))
    (kill-tasks :default)))

运行示例：

LPARALLEL-USER> (test-kill-all-tasks)
16
1
8
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.

由于创建了 10 个任务，8 个线程都被占用了。当终止掉 :default 中的任务时，所有的线程都会被终止然后重新生成（这个很费资源）。这就要避免使用 lparallel:kill-tasks 的原因。

在上述例子中，所有运行的任务都被终止了，因为这些任务都在 :default 中。假设只需要终止指定的任务，可以在创建任务时指定其 *task-category*，然后在调用 lparallel:kill-tasks 时指定这个 category。

例如，假设有两种任务：计算平方以及计算立方。分别将它们的类别设为 ’squaring-tasks 和 ‘cubing-tasks。然后随机的终止 ‘suqaring-tasks 或 ’cubing-tasks

下面是代码：

;;; kill tasks of a randomly chosen category
(defun test-kill-random-tasks ()
  (let ((channel (make-channel))
        (stream *query-io*))
    (let ((*task-category* 'squaring-tasks))
      (dotimes (i 5)
        (submit-task channel #'(lambda (x)
                                 (sleep (random 5))
                                 (format stream "~%[Squaring] ~d = ~d" x (* x x))) i)))
    (let ((*task-category* 'cubing-tasks))
      (dotimes (i 5)
        (submit-task channel #'(lambda (x)
                                 (sleep (random 5))
                                 (format stream "~%[Cubing] ~d = ~d" x (* x x x))) i)))
    (sleep 1)
    (if (evenp (random 10))
        (progn
          (print "Killing squaring tasks")
          (kill-tasks 'squaring-tasks))
        (progn
          (print "Killing cubing tasks")
          (kill-tasks 'cubing-tasks)))))

以下是执行结果：

LPARALLEL-USER> (test-kill-random-tasks)
[Cubing] 2 = 8
[Squaring] 4 = 16
[Cubing] 4
 = [Cubing] 643 = 27
"Killing squaring tasks"
4
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
[Cubing] 1 = 1
[Cubing] 0 = 0
LPARALLEL-USER> (test-kill-random-tasks)
[Squaring] 1 = 1
[Squaring] 3 = 9
"Killing cubing tasks"
5
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
[Squaring] 2 = 4
WARNING: lparallel: Replacing lost or dead worker.
WARNING: lparallel: Replacing lost or dead worker.
[Squaring] 0 = 0
[Squaring] 4 = 16

promises 和 futures

promise 和 futures 支持异步编程。

用 lparallel 的说法（lparallel-speak），lparallel:promise 就是个占位符，用来存放提供的值。 promise 对应是通过 lparallel:promise 创建，赋值的话用 lparallel:fullfill宏。

要检查 promise 有没有赋值，可以用 lparallel:fulfilledp 断言函数。最后，lparallel:force 函数用来提取 promise 中的值。注意，lparallel:forece 函数在没完成前会进行阻塞。

先看个例子来巩固下上面的知识吧：

(defun test-promise ()
  (let ((p (promise)))
    (loop
       do (if (evenp (read))
              (progn
                (fulfill p 'even-received!)
                (return))))
    (force p)))

会有以下的输出：

LPARALLEL-USER> (test-promise)
5
1
3
10
EVEN-RECEIVED!

说明：示例中会一直循环下去，直到输入的是偶数。在循环中 promise 使用 lparallel:fulfill 赋值，随后函数 lparallel:force 会强制返回这个值。

现在，来看个大点的例子吧。假设不想在那里等待 promise 赋值，而是在等待的时候做点其他的事，下面的例子中就将委托 promise 的赋值。

假设有个函数，会计算参数的平方。并且，因为这个参数，会消耗很多时间。从客户端的代码来看，只需要调用函数然后等待返回的平方值。

(defun promise-with-threads ()
  (let ((p (promise))
        (stream *query-io*)
        (n (progn
             (princ "Enter a number: ")
             (read))))
    (format t "In main function...~%")
    (bt:make-thread
     #'(lambda ()
         (sleep (random 10))
         (format stream "Inside thread... fulfilling promise~%")
         (fulfill p (* n n))))
    (bt:make-thread
     #'(lambda ()
         (loop
            when (fulfilledp p)
            do (return)
            do (progn
                 (format stream "~d~%" (random 100))
                 (sleep (* 0.01 (random 100)))))))
    (format t "Inside main function, received value: ~d~%" (force p))))

结果如下：

LPARALLEL-USER> (promise-with-threads)
Enter a number: 19
In main function...
44
59
90
34
30
76
Inside thread... fulfilling promise
Inside main function, received value: 361
NIL

说明：上面的例子没什么可说的。创建个 promise 对象 p，然后创建线程，睡眠一段随机的时间，然后在给 promise 赋值。

同时，在主线程中，创建个线程不断地检查 promise 是否被赋值。如果 promise 没有被赋值，就会打印些随机数然后继续检查。一旦 promise 被赋值，就可以在主线程中用 lparallel:force 将 promise 的值提取出来。

这表明 promise 可以被不同的线程赋值，且创建 promise 的代码不需要等待 promise 赋值完成。这点就很重要，像之前提到的， lparallel:force 是个阻塞函数。需要推迟对 promise 值的提取直到 promise 完成赋值。

还有一点需要注意就是，在使用 promise 时，一旦被赋值，调用 lparallel:force 函数返回的值总是一样的。也就是说， promise 只能赋值一次。

例如：

(defun multiple-fulfilling ()
  (let ((p (promise)))
    (dotimes (i 10)
      (fulfill p (random 100))
      (format t "~d~%" (force p)))))

输出的结果是：

LPARALLEL-USER> (multiple-fulfilling)
15
15
15
15
15
15
15
15
15
15
NIL

那么 future 与 promise 有什么区别呢？

lparallel:future 只是并行的 promise，也就是说，不会像 lparallel:promise 那样会在主线程中阻塞。future 是在自己的线程中执行（当然是在 lparallel 库中）。

future 的简单示例：

(defun test-future ()
  (let ((f (future
             (sleep (random 5))
             (print "Hello from future!"))))
    (loop
       when (fulfilledp f)
       do (return)
       do (sleep (* 0.01 (random 100)))
         (format t "~d~%" (random 100)))
    (format t "~d~%" (force f))))

结果如下：

LPARALLEL-USER> (test-future)
5
19
91
11
Hello from future!
NIL

说明：这和 promise-with-threads 的例子很像。然而，这有两个还是有不同的：第一，lparallel:future 宏有主体。这样的话 future 就能给自己赋值了。也就是说，一旦 future 的主体执行完成，lparallel:fulfilledp 对 future 对象总返回真。

第二，future 自身会通过库创建个单独的进程，所以它不会像 promise-with-threads 示例中的 promise 一样干扰当前的线程（需要指定线程给 promise 赋值，以避免阻塞）。

最有趣的一点是（即便是从 Dan Friedman 和其他人提出的实际理论来看），从概念上讲，future 是一个已赋值的 promise。也就是说，promise 就是 future 在某个时候会产生某种价值的合同，而 future 就是做这个的。

这里的意思是即便是使用 lparallel 库，future 的基本用法也是给 promise 赋值。这就意味着用户不需要创建类似 promise-with-threads 的函数。

现在来看些小例子来说明这点（必须承认，这个例子很刻意）

场景是这样的：需要读入一个数，然后计算这个数的平方。所以将这个工作从其他函数上卸下来，继续关注现有的事。当结果计算出来后，会自动将结果打印在终端，而不是要自己手动操作。

代码看起来是这样的：

;;; Callback example using promises and futures
(defun callback-promise-future-demo ()
  (let* ((p (promise))
         (stream *query-io*)
         (n (progn
              (princ "Enter a number: ")
              (read)))
         (f (future
              (sleep (random 10))
              (fulfill p (* n n))
              (force (future
                       (format stream "Square of ~d = ~d~%" n (force p)))))))
    (loop
       when (fulfilledp f)
       do (return)
       do (sleep (* 0.01 (random 100))))))

输出如下：

LPARALLEL-USER> (callback-promise-future-demo)
Enter a number: 19
Square of 19 = 361
NIL

说明：好吧，首先，创建个 promise 来生成的平方值。这就是对象 p。输入的值保存在本地变量 n 中。

然后创建个 future 对象 f。future 只计算输入值的平方然后将计算后的结果赋值给 promise。最后，需要它自动将结果打印出来，使用 force 函数提取匿名 future 的值，只输出所展示的字符串。

注意这和 Node 中的环境很像，在 Node 中，将回调函数传给其他函数，了解回调函数是在调用函数执行完后才会被调用。

最后要注意的是，下面的代码也是正常的（即便因为在单个线程上调用阻塞函数 lparallel:force）：

(force (future
(format stream "Square of ~d = ~d~%" n (force p))))

总结一下，一般用法是：需要异步计算结果保存到对象时用 promise，future 用来个这些 promise 赋值

同源： Common Lisp 和 lparallel 中的同源函数

同源函数可以说是 lparallel 库存在的原因。这些构造确实是在 lparallel 中提供了并行。但要注意，大部分构造是基于 futures 和 promises 构建的。

简要地说，同源函数只是等价于 Common Lisp 中的行数的并行函数。但是，有极少数同源函数在 Common Lisp 中没有对应的函数。

这时，了解同源函数两个基础特性比较重要：

• Constructs for fine-grained parallelism: defpun, plet, plet-if, etc.
• Explicit functions and macros for performing parallel operations -
  pmap, preduce, psort, pdotimes, etc.

can. In this post, we will focus on the second category of cognates.
第一种情况下不用准确的控制操作。主要是库会尽可能地优化和并发这种格式。本文中，关注的是第二类同源函数。

Take, for instance, the cognate function lparallel:pmap is exactly
the same as the Common Lisp equivalent, map, but it runs in
parallel. Let’s demonstrate that through an example.j
举个例子，lparallel:pmap 函数和 map 一摸一样，但前者是并行执行的。用个例子来讲解下吧。

假设有多个随机字符串，其长度是从 3 到 10，然后需要将这些字符串的长度保存到个向量中。

首先创建个 helper 函数，用来生成随机字符串：

(defvar *chars*
  (remove-duplicates
   (sort
    (loop for c across "The quick brown fox jumps over the lazy dog"
       when (alpha-char-p c)
       collect (char-downcase c))
    #'char<)))
(defun get-random-strings (&optional (count 100000))
  "generate random strings between lengths 3 and 10"
  (loop repeat count
     collect
       (concatenate 'string  (loop repeat (+ 3 (random 8))
                           collect (nth (random 26) *chars*)))))