Scala – Monadic折叠与状态monad在恒定的空间(堆和堆栈)？

是否有可能在状态monad中执行一个不断的栈和堆空间的折叠？或者是不同的功能技术更适合我的问题？

接下来的部分将介绍问题和激励用例.我正在使用Scala,但Haskell的解决方案也是受欢迎的.

折叠在邦纳多填充堆

假设斯卡拉兹7.考虑一国的monadic折叠在国家monad.为了避免堆栈溢出,我们将蹦床折叠.

import scalaz._
import Scalaz._
import scalaz.std.iterable._
import Free.Trampoline

type TrampolinedState[S,B] = StateT[Trampoline,S,B] // monad type constructor

type S = Int  // state is an integer
type M[B] = TrampolinedState[S,B] // our trampolined state monad

type R = Int  // or some other monoid

val col: Iterable[R] = largeIterableofRs() // defined elsewhere

val (count,sum): (S,R) = col.foldLeftM[M,R](Monoid[R].zero){ 
    (acc: R,x: R) => StateT[Trampoline,R] {
      s: S => Trampoline.done { 
        (s + 1,Monoid[R].append(acc,x))
      }
    }
} run 0 run

// In Scalaz 7,foldLeftM is implemented in terms of foldRight,which in turn
// is a reversed.foldLeft. This pulls the whole collection into memory and kills
// the heap.  Ignore this heap overflow. We could reimplement foldLeftM to avoid
// this overflow or use a foldRightM instead.
// Our real issue is the heap used by the unexecuted State mobits.

对于一个大的集合列,这将填满堆.

我相信,在折叠期间,为集合中的每个值(x：R参数)创建一个闭包(一个状态动作),填充堆.在执行运行0之前,不能对其进行评估,从而提供初始状态.

可以避免这个O(n)堆的使用吗？

更具体地说,可以在折叠之前提供初始状态,使得状态monad可以在每个绑定期间执行,而不是嵌套闭包以供以后评估？

还是可以建造这样的折叠,以便在国家monad运行后懒洋洋地执行？这样,下一个x：R闭包将不会被创建,直到之前的一些被评估并适合于垃圾收集.

或者这种工作有更好的功能范例吗？

应用示例

但也许我正在使用错误的工具.以下是一个示例用例的演进.我在这里走错路吗？

考虑reservoir sampling,即从一个太大的集合中一次挑选一个统一的随机k个项目,以适应内存.在Scala中,这样的功能可能是

def sample[A](col: TraversableOnce[A])(k: Int): Vector[A]

如果可以像这样使用TraversableOnce类型

val tenRandomInts = (Int.Min to Int.Max) sample 10

样品完成的工作基本上是一回事：

def sample[A](col: Traversable[A])(k: Int): Vector[A] = {
    col.foldLeft(Vector()){update(k)(_: Vector[A],_: A)}
}

但是,更新是有状态的;这取决于n,已经看到的项目数量. (它也取决于RNG,但是为了简单起见,我假设这是全局和有状态的.用于处理n的技术将会简单地扩展.那么如何处理这个状态呢？

不纯的解决方案很简单,并且使用不断的堆栈和堆栈运行.

/* Impure version of update function */
def update[A](k: Int) = new Function2[Vector[A],A,Vector[A]] {
    var n = 0
    def apply(sample: Vector[A],x: A): Vector[A] = {
        n += 1
        algorithmR(k,n,acc,x)
    }
}

def algorithmR(k: Int,n: Int,acc: Vector[A],x: A): Vector[A] = {
    if (sample.size < k) {
        sample :+ x // must keep first k elements
    } else {
        val r = rand.nextInt(n) + 1 // for simplicity,rand is global/stateful
        if (r <= k)
            sample.updated(r - 1,x) // sample is 0-index
        else
            sample
    }
}

但是纯功能解决方案呢？更新必须将n作为附加参数,并返回新值以及更新的样本.我们可以将n包含在隐式状态中,折叠累加器,例如,

(col.foldLeft ((0,Vector())) (update(k)(_: (Int,Vector[A]),_: A)))._2

但这掩盖了意图;我们只打算积累样本矢量.这个问题似乎已经准备好了,对于国家monad和一个单一的左折.让我们再试一次.

我们将使用Scalaz 7与这些导入

import scalaz._
import Scalaz._
import scalaz.std.iterable_

并运行在一个Iterable [A],因为Scalaz不支持一个可变的单体折叠.

现在定义样品

// sample using State monad
def sample[A](col: Iterable[A])(k: Int): Vector[A] = {       
    type M[B] = State[Int,B]

    // foldLeftM is implemented using foldRight,which must reverse `col`,blowing
    // the heap for large `col`.  Ignore this issue for now.
    // foldLeftM could be implemented differently or we could switch to
    // foldRightM,implemented using foldLeft.
    col.foldLeftM[M,Vector[A]](Vector())(update(k)(_: Vector[A],_: A)) eval 0
}

哪里更新

// update using State monad
def update(k: Int) = {
    (acc: Vector[A],x: A) => State[Int,Vector[A]] {
        n => (n + 1,algorithmR(k,n + 1,x)) // algR same as impure solution
    }
}

不幸的是,这将堆叠在一个大集合上.

所以让我们蹦床吧.样品现在

// sample using trampolined State monad
def sample[A](col: Iterable[A])(k: Int): Vector[A] = {
    import Free.Trampoline

    type TrampolinedState[S,B]
    type M[B] = TrampolinedState[Int,B]

    // Same caveat about foldLeftM using foldRight and blowing the heap
    // applies here.  Ignore for now. This solution blows the heap anyway;
    // let's fix that issue first.
    col.foldLeftM[M,_: A)) eval 0 run
}

哪里更新

// update using trampolined State monad
def update(k: Int) = {
    (acc: Vector[A],x: A) => StateT[Trampoline,Int,Vector[A]] {
        n => Trampoline.done { (n + 1,x) }
    }
}

这修复了堆栈溢出,但仍然为非常大的集合(或非常小的堆)打击堆.一个匿名功能
集合中的值在折叠期间创建(我相信关闭每个x：A参数),在蹦床平稳运行之前消耗堆. (FWIW,状态版本也有这个问题;堆栈溢出首先是较小的集合.)

Suspend(() => head |+| tail.foldRightM(...))

def foldM[M[_]:Monad,B](a: A,bs: Iterable[B])(f: (A,B) => M[A]): M[A] =
  if (bs.isEmpty) Monad[M].point(a)
  else Monad[M].bind(f(a,bs.head))(fax => foldM(fax,bs.tail)(f))

val MS = StateT.stateTMonadState[Int,Trampoline]
import MS._

foldM[M,R,Int](Monoid[R].zero,col) {
  (x,r) => modify(_ + 1) map (_ => Monoid[R].append(x,r))
} run 0 run