Scalaz（34）－ Free ：算法－Interpretation

我们说过自由数据结构（free structures）是表达数据类型的最简单结构。List[A]是个数据结构，它是生成A类型Monoid的最简单结构，因为我们可以用List的状态cons和Nil来分别代表Monoid的append和zero。Free[S,A]是个代表Monad的最简单数据结构，它可以把任何Functor S升格成Monad。Free的两个结构Suspend,Return分别代表了Monad的基本操作函数flatMap,point，我特别强调结构的意思是希望大家能意识到那就是内存heap上的一块空间。我们同样可以简单的把Functor视为一种算法，通过它的map函数实现运算。我们现在可以把Monad的算法flatMap用Suspend[S[Free[S,A]]来表示，那么一段由Functor
S（ADT）形成的程序（AST）可以用一串递归结构表达：Suspend(S(Suspend(S(Suspend(S(....(Return)))))))。我们可以把这样的AST看成是一串链接的内存格，每个格内存放着一个算法ADT，代表下一个运算步骤，每个格子指向下一个形成一串连续的算法，组成了一个完整的程序（AST）。最明显的分别是Free把Monad flatMap这种递归算法化解成内存数据结构，用内存地址指向代替了递归算法必须的内存堆栈（stack）。Free的Interpretation就是对存放在数据结构Suspend内的算法（ADT）进行实际运算。不同方式的Interpreter决定了这段由一连串ADT形成的AST的具体效果。

Free Interpreter的具体功能就是按存放在数据结构Suspend内的算法（ADT）进行运算后最终获取A值。这些算法的实际运算可能会产生副作用，比如IO算法的具体操作。scalaz是通过几个运算函数来提供Free Interpreter，包括：fold,foldMap,foldRun,runFC,runM。我们先看看这几个函数的源代码：

/** Catamorphism. Run the first given function if Return, otherwise, the second given function. */
final def fold[B](r: A => B, s: S[Free[S, A]] => B)(implicit S: Functor[S]): B =
resume.fold(s, r)

/**
* Catamorphism for `Free`.
* Runs to completion, mapping the suspension with the given transformation at each step and
* accumulating into the monad `M`.
*/
final def foldMap[M[_]](f: S ~> M)(implicit S: Functor[S], M: Monad[M]): M[A] =
this.resume match {
case -\/(s) => Monad[M].bind(f(s))(_.foldMap(f))
case \/-(r) => Monad[M].pure(r)
}

/** Runs to completion, allowing the resumption function to thread an arbitrary state of type `B`. */
final def foldRun[B](b: B)(f: (B, S[Free[S, A]]) => (B, Free[S, A]))(implicit S: Functor[S]): (B, A) = {
@tailrec def foldRun2(t: Free[S, A], z: B): (B, A) = t.resume match {
case -\/(s) =>
val (b1, s1) = f(z, s)
foldRun2(s1, b1)
case \/-(r) => (z, r)
}
foldRun2(this, b)
}

/**
* Runs to completion, using a function that maps the resumption from `S` to a monad `M`.
* @since 7.0.1
*/
final def runM[M[_]](f: S[Free[S, A]] => M[Free[S, A]])(implicit S: Functor[S], M: Monad[M]): M[A] = {
def runM2(t: Free[S, A]): M[A] = t.resume match {
case -\/(s) => Monad[M].bind(f(s))(runM2)
case \/-(r) => Monad[M].pure(r)
}
runM2(this)
}

/** Interpret a free monad over a free functor of `S` via natural transformation to monad `M`. */
def runFC[S[_], M[_], A](sa: FreeC[S, A])(interp: S ~> M)(implicit M: Monad[M]): M[A] =
sa.foldMap[M](new (({type λ[α] = Coyoneda[S, α]})#λ ~> M) {
def apply[A](cy: Coyoneda[S, A]): M[A] =
M.map(interp(cy.fi))(cy.k)
})

我们应该可以看出Interpreter的基本原理就是把不可运算的抽象指令ADT转换成可运算的表达式。在这个转换过程中产生运算结果。我们下面用具体例子一个一个介绍这几个函数的用法。还是用上期的例子：

object qz {
sealed trait Quiz[+Next]
object Quiz {
//问题que:String, 等待String 然后转成数字或操作符号
case class Question[Next](que: String, n: String => Next) extends Quiz[Next]
case class Answer[Next](ans: String, n: Next) extends Quiz[Next]
implicit object QFunctor extends Functor[Quiz] {
def map[A,B](qa: Quiz[A])(f: A => B): Quiz[B] =
qa match {
case q: Question[A] => Question(q.que, q.n andThen f)
case Answer(a,n) => Answer(a,f(n))
}
}
//操作帮助方法helper methods
def askNumber(q: String) = Question(q, (inputString => inputString.toInt))  //_.toInt
def askOperator(q: String) = Question(q, (inputString => inputString.head.toUpper.toChar)) //_.head.toUpper.toChar
def answer(fnum: Int, snum: Int, opr: Char) = {
def result =
opr match {
case 'A' => fnum + snum
case 'M' => fnum * snum
case 'D' => fnum / snum
case 'S' => fnum - snum
}
Answer("my answer is: " + result.toString,())
}
implicit def quizToFree[A](qz: Quiz[A]): Free[Quiz,A] = Free.liftF(qz)
}
import Quiz._
val prg = for {
fn <- askNumber("The first number is:")
sn <- askNumber("The second number is:")
op <- askOperator("The operation is:")
_ <- answer(fn,sn,op)
} yield()

prg是一段功能描述：在提示后读取一个数字，重复一次，再读取一个字串，把读取的数字和字串用来做个运算。至于怎么提示、如何读取输入、如何运算输入内容，可能会有种种不同的方式，那要看Interpreter具体是怎么做的了。好了，现在我们看看如何用fold来运算prg：fold需要两个入参数：r:A=>B,一个在运算终止Return状态时运行的函数，另一个是s:S[Free[S,A]]=>B,这个函数在Suspend状态时运算入参数ADT：

def runQuiz[A](p: Free[Quiz,A]): Unit= p.fold(_ => (), {
case Question(q,f) => {
println(q)
runQuiz(f(readLine))
}
case Answer(a,n) => println(a)
})

注意runQuiz是个递归函数。在Suspend Question状态下，运算f(readLine)产生下一个运算。在这个函数里我们赋予了提示、读取正真的意义，它们都是通过IO操作println,readLine实现的。

object main extends App {
import freeRun._
import qz._
runQuiz(prg)
}

运行结果：

The first number is:
3
The second number is:
8
The operation is:
mul
my answer is: 24

结果正是我们期待的。但这个fold方法每调用一次只运算一个ADT，所以使用了递归算法连续约化Suspend直到Return。递归算法很容易造成堆栈溢出异常，不安全。下一个试试foldMap。foldMap使用了Monad.bind连续通过高阶类型转换（natural transformation）将ADT转换成运行指令，并在转换过程中实施运算：

object QuizConsole extends (Quiz ~> Id) {
import Quiz._
def apply[A](qz: Quiz[A]): Id[A] = qz match {
case Question(a,f) => {
println(a)
f(readLine)
}
case Answer(a,n) => println(a);n
}
}
//运行foldMap
prg.foldMap(QuizConsole)
//结果一致

上面的natural transformation是把Quiz类型转成Id类型。Id[A]=A，所以高阶类型Quiz可以被转换成基本类型Unit（println返回Unit）。这个例子同样用IO函数来实现AST功能。我们也可以用一个模拟的输入输出方式来测试AST功能，也就是用另一个Interpreter来运算AST，我们可以用Map[String,String]来模拟输入输出环境：

type Tester[A] = Map[String, String] => (List[String], A)
object QuizTester extends (Quiz ~> Tester) {
def apply[A](qa: Quiz[A]): Tester[A] = qa match {
case Question(q,f) => m => (List(),f(m(q)))
case Answer(a,n) => m => (List(a),n)
}
}
implicit object testerMonad extends Monad[Tester] {
def point[A](a: => A) = _ => (List(),a)
def bind[A,B](ta: Tester[A])(f: A => Tester[B]): Tester[B] =
m => {
val (o1,a) = ta(m)
val (o2,b) = f(a)(m)
(o1 ++ o2, b)
}
}

Tester必须是个Monad，所以我们必须提供隐式对象testerMonad。看看运算结果：

val m = Map(
"The first number is:" -> "8",
"The second number is:" -> "3",
"The operation is:" -> "Sub"
)
println(prg.foldMap(QuizTester).apply(m))
//(List(my answer is: 5),())

foldRun通过入参数f:(B,S[Free[S,A]])=>(B,Free[S,A])支持状态跟踪,入参数b:B是状态初始值。我们先实现这个f函数：

type FreeQuiz[A] = Free[Quiz,A]
def quizst(track: List[String], prg: Quiz[FreeQuiz[Unit]]): (List[String], FreeQuiz[Unit]) =
prg match {
case Question(q,f) => {
println(q)
val input = readLine
(q+input :: track, f(input))
}
case Answer(a,n) => println(a); (a :: track, n)
}

运行foldRun的结果如下：

println(prg.foldRun(List[String]())(quizst)._1)
The first number is:
2
The second number is:
4
The operation is:
Mul
my answer is: 8
List(my answer is: 8, The operation is:Mul, The second number is:4, The first number is:2)

下一个是runM了，它的入参数就是一个S[_]到M[_]的转换函数：f: S[Free[S,A]]=>M[Free[S,A]]。我们先实现了这个f函数：

type FreeQuiz[A] = Free[Quiz,A]
def runquiz[A](prg: Quiz[FreeQuiz[A]]): Id[FreeQuiz[A]] =
prg match {
case Question(q,f) => {
println(q)
f(readLine)
}
case Answer(a,n) => println(a); n
}

测试运行runM：

prg.runM(run quiz)
The first number is:
4
The second number is:
2
The operation is:
Mul
my answer is: 8

我们曾经介绍过有些F[_]是无法实现map函数的，因此无法成为Functor，如以下ADT：

sealed trait Calc[+A]
object Calc {
case class Push(value: Int) extends Calc[Unit]
case class Add() extends Calc[Unit]
case class Mul() extends Calc[Unit]
case class Div() extends Calc[Unit]
case class Sub() extends Calc[Unit]
implicit def calcToFree[A](ca: Calc[A]) = Free.liftFC(ca)
}
import Calc._
val ast = for {
_ <- Push(23)
_ <- Push(3)
_ <- Add()
_ <- Push(5)
_ <- Mul()
} yield ()                                        //> ast  : scalaz.Free[[x]scalaz.Coyoneda[Exercises.interact.Calc,x],Unit] = Gosub()

从Calc无法获取B类型值，所以无法实现Calc.map，因而Calc无法成为Functor。runFC就是专门为运算Calc这样的非Functor高阶类型值的。runFC需要一个FreeC[S,A]类型入参数:

/** A free monad over the free functor generated by `S` */
type FreeC[S[_], A] = Free[({type f[x] = Coyoneda[S, x]})#f, A]
}

可以得出runFC是专门为Coyoneda设计的。Coyoneda可以替代Calc[A]，又是一个Functor，所以可以用Free产生Calc类型的Monad。我们先把Interpreter实现了：

type Stack = List[Int]
type StackState[A] = State[Stack,A]
object CalcStack extends (Calc ~> StackState) {
def apply[A](ca: Calc[A]): StackState[A] = ca match {
case Push(v) => State((s: Stack) => (v :: s, ()))
case Add() => State((s: Stack) => {
val a :: b :: t = s
((a+b) :: t,())
})
case Mul() => State((s: Stack) => {
val a :: b :: t = s
((a * b) :: t, ())
})
case Div() => State((s: Stack) => {
val a :: b :: t = s
((a / b) :: t,())
})
case Sub() => State((s: Stack) => {
val a :: b :: t = s
((a - b) :: s, ())
})
}
}

这个Interpreter用的是Stack内元素操作的运算方式。用runFC对ast运算的结果：

println(Free.runFC(ast)(CalcStack).apply(List[Int]()))
//(List(130),())

以上示范了针对任何抽象的Monadic Programm,我们如何通过各种Interpreter的具体实现方式来确定程序功能的。