讓我們使用 Scala。
我試圖找到最好的方法來對迭代器的一些元素進行機會性、部分和異步的預計算，否則這些元素將被同步處理。
下圖說明了問題。
有一個主執行緒（藍色）接受一個迭代器和一個狀態。狀態包含必須防止并發訪問的可變資料。此外，狀態必須在迭代器從頭開始按順序處理時更新，因為迭代器的元素依賴于先前的元素。此外，依賴的性質是事先不知道的。
Processing some elements may lead to substantial overhead (2 orders of magnitude) compared to others, meaning that some elements are 1ms to compute and some elements are 300ms to compute. It would lead to significant improvements in terms of running time if I could pre-process the next k elements speculatively. A speculative pre-processing on asynchronous threads is possible (while the blue thread is synchronously processing), but the pre-processed data must be validated by the blue thread, whether the result of pre-computation is valid at that time. Usually (90% of the time), it should be valid. Thus, launching separate asynchronous threads to pre-process the remaining portion of the iterator speculatively would spear many 300s of milliseconds in running time.
I have studied comparisons of asynchronous and functional libraries of Scala to understand better which model of computation, or in other words, which description of computation (which library) could be a better fit to this processing problem. I was thinking about communication patterns and came about with the following ideas:
- AKKA
Use an AKKA actor Blue for the blue thread that takes the iterator, and for each step, it sends a Step message to itself. On a Step message, before it starts the processing of the next ith element, it sends a PleasePreprocess(i k) message with the i kth element to one of the k pre-processor actors in place. The Blue would Step to i 1 only and only if PreprocessingKindlyDone(i 1) is received.
- AKKA Streams
AFAIK AKKA streams also support the previous two-way backpressure mechanism, therefore, it could be a good candidate to implement what actors do without actually using actors.
- Scala Futures
While the blue thread processes elements ˙processElement(e)˙ in iterator.map(processElement(_)), then it would also spawn Futures for preprocessing. However, maintaining these pre-processing Futures and awaiting their states would require a semi-blocking implementation in pure Scala as I see, so I would not go with this direction to the best of my current knowledge.
- Use Monix
I have some knowledge of Monix but could not wrap my head around how this problem could be elegantly solved with Monix. I'm not seeing how the blue thread could wait for the result of i 1 and then continue. For this, I was thinking of using something like a sliding window with foldLeft(blueThreadAsZero){ (blue, preProc1, preProc2, notYetPreProc) => ... }, but could not find a similar construction.
可能有一些我沒有提到的庫可以更好地表達計算模式。
我希望我已經充分描述了我的問題。感謝您的提示/想法或代碼片段！
uj5u.com熱心網友回復：
如果你的藍色執行緒恰好比黃色執行緒快，你無論如何都需要阻塞。我認為您不需要任何花哨的庫，“vanilla scala”應該這樣做（就像在大多數情況下實際上一樣）。像這樣的東西，也許...
def doit[T,R](it: Iterator[T], yellow: T => R, blue: R => R): Future[Seq[R]] = it
.map { elem => Future(yellow(elem)) }
.foldLeft(Future.successful(List.empty[R])) { (last, next) =>
last.flatMap { acc => next.map(blue).map(_ :: acc) }
}.map(_.reverse)
我沒有測驗或編譯它，所以它可能需要一些調整，但從概念上講，這應該可行：通過迭代器并立即開始預處理，然后折疊以按順序在每個完成的預處理上折疊“驗證”。
uj5u.com熱心網友回復：
我會將處理分為兩個步驟，一個可以并行運行的預處理，另一個必須是串行的。
然后，您可以從迭代器創建資料流，執行并行映射，應用預處理步驟并完成fold
我個人會使用fs2，但同樣的方法可以用任何流媒體解決方案來表達，比如AkkaStreams、Monix Observables或ZIO ZStreams
import fs2.Stream
import cats.effect.IO
val finalState =
Stream
.fromIterator[IO](iterator = ???, chunkSize = ???)
.parEvalMap(elem => IO(preProcess(elem))
.compile
.fold(initialState) {
case (acc, elem) =>
computeNewState(acc, elem)
}
PS：請記住進行基準測驗以確保并行性實際上加快了速度；這可能不值得麻煩。
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/450229.html