π，序曲，第一个reducer

摘要：在和的参数列表里，顺序不重要，这是并发本质对于只请求一个的情况，和没有区别。在并发编程里之需要并发组合这一种操作符，不需要再发明一个顺序组合操作符号，因为它只是并发组合的一个特例。

花了很久的时间学习π calculus；天资愚钝至今尚未学明白，好在不影响写代码。

任何一种和计算或者编程相关的数学理论，都可以有两种不同的出发点：一种是可以作为基础理论（或计算模型）解释程序员们每天用各种语言写下的代码，它背后的本质是怎样的；这就象用物理学解释生活里看到的各种自然现象；另一种是通过对理论的学习，了解到它在概念层面上具体解决了什么问题，以及针对哪类问题特别有效，在编程开发实践中尝试应用其思想。

后一种相对玄学，但是反过来说这个思考和实践的过程对理解理论很有帮助。

π和λ一样很抽象，离编程实践很远，而且，完全存在可能性，一个完整的实践需要语言和运行环境一级支持。但是学习一件事物呢，不要太功利，找到乐趣开动思维是最重要的，在能真正在工程上大面积应用之前，不妨就把它看作是一个益智游戏。这样的心态就会让学习变得富有乐趣，不容易焦虑或者有挫折感。

我不打算从符号入手讲解π，但是它的基础概念要交代一下。

π是关于进程的算术（或者叫演算）；算术（Calculus）一词不如想象的那么吓人，不要因为曾经噩梦般的考试生活对它天生恐惧。算术的意思只是说，我们希望我们的代码里的构件，类也好，方法也好，他们是可以如此灵活的组合使用的，就像我们在数学上的运算符，可以算整数、自然数、复数、向量、矩阵、张量、等等；数学上有很多的运算符可用，大多数运算符都能应用在相当广泛的数学对象上；所以我们说数学系统是丰富的，是强大的思维工具和解决问题的方法。

说π是进程算术的意思很自然，就是构建一个系统时把它看作是很多进程的组合；在这里进程的含义和我们在代码中写下的函数差不多，但是它不是指操作系统意义上的进程，也不像λ那样可以描述纯函数。

除了过程，π里只有一个概念：通讯。构成系统的多个进程，包括大量实际系统中的动态过程，他们用通讯的方式交互；这两者就构成了系统的全部。

π里的通讯和Golang或者CSP里的channel，或者，Alan Kay定义的那种OO或者Actor Model里的message，又或者，我们实际在编程中使用的socket或者ipc，有没有关系？关系肯定是有的，但是π里定义的通讯比所有这些都更加纯粹；而且，在π里只有通讯这一件事；这预示着，在这个系统里的所有行为，都由通讯来完成。

我们来看一下π里最基础也是最重要的一个表达式：

|

（这个表达式在segmentfault的显示有误，应该是一行，中间用vertical pipe，在π里表示并发组合）

在|左侧的表达式的意思是，有一个叫做c的通讯通道，可以收到一个值，收到这个值之后P才可以开始演算（估值），P里面的x，都替换成收到的值；当然这个值是个常数是我们最喜闻乐见的，但实际上也可能收到一个完整的π表达式（就成了High Order了）。

在右侧的表达式和左侧相反，它指的是P过程如果要开始演算，前提条件是向通讯通道c发送一个y出去；这个从程序员的角度看感觉可能没法理解，console.log()之后才能继续执行是什么意思？好像从来没有遇到过输出阻塞程序运行而且让程序员伤脑筋的事儿。

但是这个表达式在π里很重要；在编程里同样很重要。

输出前缀在π里表述的意思是一个过程被blocking到有请求时才开始。比如实现一个readable stream，在buffer里的数据枯竭或者低于警戒线的时候才会启动代码读取更多数据填充buffer。

而前面这个表达式，可以看作是没有buffer的两个过程，一个读，一个写；然后两侧的过程都可以开始执行，而且，是以并发的方式。在π里，或者其他类似的符号系统里，这种表达式变换叫做reduction，和数学表达式销项简化是一样的。

所以我们写下的第一个玩具级代码片段里，这个类的名字就叫做Reducer。

Reducer可以接受一个callback形式的函数作为生产者（producer），producer等待到reducer对象的on方法被调用时开始执行，当它产生结果时更新reducer对象的error或者data成员，同时，等待这个值的函数（在调用on时被保存在consumers成员数组中，被全部调用。

这个producer只能运行一次，如果完成之后还有on请求，会同步调用请求函数。只工作一次这个限制让这个类无法做到可变更数据的观察，不过那不是我们现在需要考虑的问题。

class Reducer {
  constructor (producer) {
    if (typeof producer !== "function") throw new Error("producer must be a function")
    this.producer = producer
  }

  on (f) {
    if (Object.prototype.hasOwnProperty.call(this, "data") ||
      Object.prototype.hasOwnProperty.call(this, "error")) {
      f() 
    } else {
      if (this.consumers) {
        this.consumers.push(f)
      } else {
        this.consumers = [f] 
        this.producer((err, data) => {
          if (err) {
            this.error = err 
          } else {
            this.data = data
          }   
          const consumers = this.consumers
          delete this.producer
          delete this.consumers
          consumers.forEach(f => f())
        })  
      }   
    }   
  }
}

那么你可能会问，node.js里有emitter了，还有各种stream，为什么要多带带写这样一个Reducer？

在成品的开发框架中提供的类，一般都是完善的工具，它包含的不只有一个概念，而且要应对很多实际的使用需求。

而我们这里更强调概念，这是第一个原因；第二个原因，是reducer更原始（primitive)，它不是用于继承的，也没有定义任何事件名称，即，它没有行为语义。

node.js里的emitter可以在π的意义上看作一个表达式，每一个类似write之类的方法都是一个通讯channel，每一个on的事件名称也是一个通讯channel，换句话说，它不是一个基础表达式。

把一个非基础表达式作为一个基础构件是设计问题，当我们需要表达它没有提供的更基础或者更灵活的语义要求时就有麻烦，比如我们有两个event source其中一个出错时：

  const src1onData = data => { ... }
  const src1onError = err => {
    src1.removeListener("data", src1onData)
    src1.removeListener("error", src1onError)
    src1.on("error", () => {})  // mute further error
    src2.removeListener("data", src2onData)
    src2.removeListener("error", src2onError)
    src2.on("error", () => {})  // mute further error
    src1.destroy()
    src2.destroy()
    callback(err)
  }

  const src2onData = data => { ... }
  const src2onError = err => {
    ....
  }

  source1.on("data", src1onData)
  source1.on("error", src1onError)
  source2.on("data", src2onData)
  source2.on("error", src1onError) 

在node.js里类似这样的代码不在少数；造成这个困难的原因，就是“互斥”这个在π里只要一个加号（+）表示的操作，在emitter里受到了限制；而且emitter的代码已经有点重了，自己重载不是很容易。

在看实际使用代码之前来看一点小小的算术逻辑。

// one finished
const some = (...rs) => {
  let next = rs.pop() 
  let fired = false
  let f = x => !fired && (fired = true, next(x))
  rs.forEach(r => r.on(f))
}

// all finished
const every = (...rs) => {
  let next = rs.pop()
  let arr = rs.map(r => undefined)
  let count = rs.length 
  rs.forEach((r, i) => r.on(x => (arr[i] = x, (!--count) && next(...arr))))
}

module.exports = {
  reducer: f => new Reducer(f),
  some,
  every,
}

就像javascript的数组方法一样，我们希望能够灵活表达针对一组reducer的操作。比如第一个some方法；它用了javascript的rest parameters特性，参数中最后一个是函数，其他的都是reducer，这样使用代码的形式最好读。

some的意思是同时on多个reducer，但只要有一个有值了，最后一个参数函数就被调用。

every的意思也是同时on多个reducer，但需要全部有值，才会继续。

这里的代码很原始，而且对资源不友好，但用于说明概念可以了。

最后来看一点实际使用的代码：

// bluetooth addr (from ssh)
const baddr = reducer(callback => getBlueAddr(ip, callback)) 
// bluetooth device info
const binfo = reducer(callback => 
  pi.every(baddr, () => ble.deviceInfo(baddr.data, callback)))

第一个reducer是baddr是取设备蓝牙地址的；getBlueAddr是很简单还是很复杂没关系。这句话说明读取baddr在当前上下文下没有其他依赖性，可以直接执行；但是这个语句并没有立刻开始读取蓝牙地址的过程。它相当于我们前面写的π表达式：

即过程P（getBlueAddr）能产生（输出）一个蓝牙地址，但是它会一直等到有人来读的时候才会开始运行。

出发这个过程开始执行的代码在在最后一句，在binfo的producer里。这个pi.every(...)的调用，就相当于：

因为这个代码在binfo的producer里，所以它还没开始执行，也不会和baddr的producer发生reduction。

在binfo的producer代码里出现了对另一个reducer的on, pi.every, pi.some之类的操作，就直接表述了binfo对baddr的依赖关系。这是这种看起来有点小题大作的写法的一个好处，就是你阅读代码时依赖性是一目了然。

这两行代码在运行后，两个producer过程都没开始，因为没有一个reducer被on了。如果你需要触发这个过程，可以写：

pi.every(binfo, () => {
  console.log("test every", baddr.data, binfo.data)
})

当然这个写法在并发编程里不推荐，因为你是读了binfo的代码知道依赖性的，否则console.log可能会发生错误。推荐的做法是一股脑把你要的reducer都写到every或some里去，他们之间的依赖性对every或者some的回调函数来说是黑盒的：

pi.every(baddr, binfo, () => {
  console.log("test every", baddr.data, binfo.data)
})

无论是some还是every，都是让所有被请求的reducer的producers同时开始工作，即并发组合。在every和some的参数列表里，顺序不重要，这是并发本质；对于只请求一个reducer的情况，every和some没有区别。

如果你需要顺序组合，大概可以这样写：

pi.every(baddr, () => pi.every(binfo, () => {
  ...
}))

不过为什么会需要顺序呢？我们在写流程代码的时候需要的，不是顺序，是依赖性；偶尔发生的完全没有数据传递的顺序，比如另一个读取文件的过程必须等到一个写入文件的过程结束，也可以理解为前面一个过程产生了一个null结果是后面们一个过程需要的。

上面这句话是Robin Milner在他的图灵奖获奖发言里说的。在并发编程里之需要并发组合这一种操作符，不需要再发明一个顺序组合操作符号，因为它只是并发组合的一个特例。

在node.js里，因为异步特性，分号（;）是语言意义上的顺序组合，但是模型意义上的并发组合。callback, emitter, promise，async/await，以及上面的这个形同柯里化的pi.every语句，都是顺序组合的表达。但是我相信你看完这篇文章后会理解，在并发编程里，只有局部是为了便于书写需要这种顺序组合。
并发编程和顺序编程的本质不同，是前者在表达依赖性，而不是顺序。

我鼓励你用Reducer写点实际的代码，虽然它不能应对连续变化的值，只是单发（one-shot）操作，但很多时候也是可以的，比如写流程，或者写http请求。

而说道写流程，我不得不说π的一大神奇特性，就是它的通讯语义已经足够表达所有流程。就像你在这里看到的代码一样，事实上用π可以构件整个程序表达顺序。

事实上我在最近几周就在写测试代码。有大量的set up/tear down和各种通讯。不同的测试配置。用π写出来的代码我最终不关心每个测试下如何做不同的初始化，因为代码全部是Lazy的，我只要在最后用every一次性Pull所有我要的reducer即可。

至于执行顺序，老实说我也不晓得。这就是并发编程！

这里有一点rx的味道对吗？

不过我不熟悉rx，我需要的也不是数据流模型；我关注的是过程的组合，如何清晰的看出依赖性，如何优雅的处理错误。

这里写的Reducer非常有潜力，它体现在：

你看到了every和some，实际上我们可以做很多复杂的逻辑在里面，比如第一个错误，比如错误类型的过滤器，比如收集够指定数量的结果就返回；

分开错误处理和成功的代码路径是可能的，Reducer里可以只on错误结果，或者正确结果；

而最重要的rx的不同，是reducer里可以装入比简单的callback更rich的函数或者对象，例如有cancel方法的，能emit progress事件的，等等；

前面说过，π里有一个+号表示互斥过程；象some或者every一样写一个互斥的on多个reducer，很容易；

互斥的一个较为复杂的情况是conditional的，这个其实也很容易写，相当于reducer级联了，写在前面的用于条件估值；更复杂的情况的是pattern matching，即用pattern选择继续执行的过程，那就更帅了，用库克的话说，I am thrilled；

All in all，还是那句老话，less is more。Emitter的设计错误在于它的目的是提供继承，而不是用于实现灵活的代数方法。

当然，reducer也只是刚刚开始。几个月后，我会再回来的。

补：文中所述的最基础的π的reduction的严格表述如下，左侧的name z从channel x出去后被vertical pipe右侧接收到，Q表达式里的y因此全部替换成z，[z/y]用于表述这个替换，称为alpha-conversion，而这个表达式从左侧到右侧的变换，就是beta-reduction。