一定のスタックとヒープ空間で State モナドで折り畳みを実行することは可能ですか? または、別の機能的手法が私の問題により適していますか?
次のセクションでは、問題と動機付けとなるユース ケースについて説明します。私は Scala を使用していますが、Haskell でのソリューションも歓迎します。
Stateモナドのフォールドがヒープを埋める
Scalaz 7 を仮定します。 State モナドのモナド折り畳みを考えます。スタック オーバーフローを避けるために、フォールドをトランポリンします。
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, S, 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.
大規模なコレクションのcol場合、これでヒープがいっぱいになります。
フォールド中に、コレクション (パラメーター) 内の値ごとにクロージャー (State mobit) が作成されx: R、ヒープがいっぱいになると思います。が実行されるまで、これらはどれも評価できずrun 0、初期状態が提供されます。
この O(n) ヒープの使用を回避できますか?
より具体的には、後で評価するためにクロージャをネストするのではなく、State モナドが各バインド中に実行できるように、フォールドの前に初期状態を提供できますか?
それとも、State モナドの後に遅延実行されるように折り畳みを構築できrunますか? このようにして、次のx: Rクロージャーは、前のクロージャーが評価されてガベージ コレクションに適したものになるまで作成されません。
それとも、この種の作業のためのより優れた機能パラダイムはありますか?
適用例
しかし、おそらく私は仕事に間違ったツールを使用しています。ユースケースの例の進化は次のとおりです。私はここで間違った道をさまよっていますか？
リザーバー サンプリング、つまり、k大きすぎてメモリに収まらないコレクションから均一なランダム アイテムを1 回のパスで選択することを検討してください。Scala では、そのような関数は次のようになります。
def sample[A](col: TraversableOnce[A])(k: Int): Vector[A]
そして、タイプにポンピングされた場合、TraversableOnceこのように使用できます
val tenRandomInts = (Int.Min to Int.Max) sample 10
によって行われる作業sampleは、基本的に次のfoldとおりです。
def sample[A](col: Traversable[A])(k: Int): Vector[A] = {
col.foldLeft(Vector()){update(k)(_: Vector[A], _: A)}
}
ただし、updateステートフルです。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
}
}
しかし、純粋に機能的なソリューションはどうでしょうか? update追加のパラメーターとして取りn、更新されたサンプルと共に新しい値を返す必要があります。n暗黙的な状態にフォールド アキュムレータを含めることができます。
(col.foldLeft ((0, Vector())) (update(k)(_: (Int, Vector[A]), _: A)))._2
しかし、それは意図をあいまいにします。サンプルベクトルを累積するだけです。この問題は、State モナドとモナドの左折畳みのために用意されているようです。もう一度試してみましょう。
これらのインポートで Scalaz 7 を使用します
import scalaz._
import Scalaz._
import scalaz.std.iterable_
Iterable[A]Scalaz は a のモナド畳み込みをサポートしていないため、an を操作しTraversableます。
sample現在定義されています
// 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, acc, x)) // algR same as impure solution
}
}
残念ながら、これは大規模なコレクションのスタックを吹き飛ばします。
それではトランポリンしましょう。sample今でしょ
// sample using trampolined State monad
def sample[A](col: Iterable[A])(k: Int): Vector[A] = {
import Free.Trampoline
type TrampolinedState[S, B] = StateT[Trampoline, 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, Vector[A]](Vector())(update(k)(_: Vector[A], _: 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, algorithmR(k, n + 1, acc, x) }
}
}
これにより、スタック オーバーフローが修正されますが、非常に大きなコレクション (または非常に小さなヒープ) のヒープは依然として吹き飛ばされます。コレクション内の値ごとに 1 つの匿名関数がフォールド中に作成され (各パラメーターを閉じると思いx: Aます)、トランポリンが実行される前にヒープを消費します。(FWIW、State バージョンにもこの問題があります。スタック オーバーフローは、小さなコレクションで最初に表面化するだけです。)