Is there an easy way to handle threads when using mutually recursive functions?

Question

I translated the incoming stream of objects into a tree data structure, using different classes to represent different types of nodes in the tree. To process threads, I have a number of mutually recursive functions that will branch depending on the next object in the stream, call each other to do part of the processing, and finally create a single node in the tree; in other words, it is a more or less recursive descent parser.

I am currently using Iterator to represent a stream of objects. Iterator's consistency makes it relatively easy to work, since each function only has to step forward along the iterator forward, and all other functions will automatically continue where the previous function stopped, without having to explicitly track the current position in the stream.

However, the combination of the need to use different viewing functions, as well as the desire to make the analyzer more functional and make more efficient use of the Scala template matching functions, has recently made this problem very difficult from the point of view of state c, as I, for example, must track elements already consumed in order to implement lookahead.

As a potential replacement, I considered the more functional Stream class in Scala; however, using Stream instead requires that I explicitly track the current position so that other functions are read from the correct part of the stream. I experimented with the fact that each function takes a stream to read from a parameter and returns the part of the stream that has not yet been processed as part of the tuple, along with the actual return value, for example:

def f2(s: Stream[X])() = {
  val a #:: s2 = s
  val (b, s3) = a match {
    case A => f2(s2)() match { case (x, s) => (Some(x), s) }
    case _ => (None, s2)
  }
  val (c, s4) = f3(s3)(1, 2, 3)
  val (d, s5) = f4(s4)()
  val e #:: s6 = s5
  (new MyClass(a, b, c, d, e), s6)
}

This, however, is incredibly cumbersome to use, that with all the additional dummy variables that I need to create and track, just pass the rest of the stream to the next function, and it needs to be constantly unpacked. Is there an easier way to handle threads or, alternatively, a better way to get closer to the problem as a whole?

Answer 1

, play-iteratees scalaz-stream, , .

:

• Elem

• , , , Result, -

  type State[Elem, Result] = Elem => (State[Elem,Result], Option[Result])
  

• Stream[Elem] => Stream[Result] .

, :

 class PStateM[Elem, Result](val receiveM: Option[Elem] => (Option[PStateM[Elem, Result]], Option[Result]))

P , M Monadic, monad - Option ( , Future teratee)

Option 2.

Option[Elem] , , None .

Option[PStateM[Elem, Result]] , .

, 2.:

  class PState[Elem, Result](val receive: Elem => (PStateM[Elem, Result], Option[Result])) extends PStateM[Elem, Result]({
    case Some(elem) => receive(elem) match {case (next, res) => (Some(next), res)}
    case None => (None, None)
  })

Nest , 3:

  def process[Elem, Result](state: PStateM[Elem, Result], source: Stream[Elem]): Stream[Result] = source match {
    case Stream() => state.receiveM(None)._2.toStream
    case head #:: tail => {
      val (next, res) = state.receiveM(Some(head))
      next match {
        case None => res.toStream
        case Some(nState) => res match {
          case None => process(nState, tail)
          case Some(res) => res #:: process(nState, tail)
        }
      }
    }
  }

, match es, . , . :

  case class MapS[A, B](f: A => B) extends PState[A, B](elem => (MapS(f), Some(f(elem))))

  case class FoldS[A, B](z: B, f: (B, A) => B) extends PStateM[A, B]({
    case Some(elem) => (Some(FoldS(f(z, elem), f)), None)
    case None => (None, Some(z))
  })

  case class FilterS[A](f: A => Boolean) extends PState[A, A](elem => (FilterS(f), Some(elem) filter f))

,

println(process(MapS[Int, Int](_ * 2), Stream.from(1)).take(10).toList)
println(process(FilterS[Int](_ % 2 == 1), Stream.from(1)).take(10).toList)
println(process(FoldS[Int, Long](0L, _ + _.toLong), 1 to 100000 toStream))

10 , 10 StackOverflowError

, . , . , , :

• node
• count children
• node - count
• 1

,

  trait Node
  case class Branch(elem: Int, children: IndexedSeq[Int]) extends Node
  case class Leaf(elem: Int) extends Node

:

case class NodeElem() extends PState[Int, Node](elem => (NodeChildCnt(elem), None))

case class NodeChildCnt(elem: Int) extends PState[Int, Node]({
  case 0 => (NodeElem(), Some(Leaf(elem)))
  case count => (NodeChildElem(elem, count, IndexedSeq.empty), None)
})

case class NodeChildElem(elem: Int, count: Int, children: IndexedSeq[Int]) extends PStateM[Int, Node]({
  case Some(child) =>
    if (count > 1)
      (Some(NodeChildElem(elem, count - 1, children :+ child)), None)
    else (Some(NodeElem()), Some(Branch(elem, children :+ child)))
  case None => (None, Some(Branch(elem, children)))

})

: NodeElem node , NodeChildCnt Leaf, 0 NodeChildElem , .

println(process(NodeElem, "1 2 2 3 2 2 4 5 3 0 4 0 5 4 6 7".split(' ').toStream.map(_.toInt)).toList)

5'th node:

List(Branch(1,Vector(2, 3)), Branch(2,Vector(4, 5)), Leaf(3), Leaf(4), Branch(5,Vector(6, 7)))