Better topology resolution (#89)

* Better topology resolution

* error example

* par added

* revert

* Test fixed

aqua-src/error.aqua

@@ -1,23 +1,15 @@
 service Demo("demo"):
 	get4: u64, u64, string -> u64
+	bool: -> bool
+	log: string, bool -> ()
 
-const bbb = 5
-const bbb ?= 2
-const ccc = "privet"
-const ddd = ccc
-const eee = bbb
-const asd = eee
--- const ttt = -2
-
-func two(variable: u64) -> u64:
-    v <- Demo.get4(variable, eee, ddd)
-    -- if bbb == ttt:
-    --     Demo.get4(variable, eee, ddd)
-    -- else:
-    --     Demo.get4(variable, eee, ddd)
-    <- variable
-
-func three() -> u64:
-    variable <- Demo.get4(asd, eee, ddd)
-    res <- two(variable)
-    <- variable
+func initAfterJoin(users: []string) -> []string:
+    for user <- users par:
+        on user:
+            isOnline <- Demo.bool()
+        if isOnline:
+            on user via "user relay":
+                Demo.log("me", true)
+        Demo.log(user, isOnline)
+        par Demo.log(user, isOnline)
+    <- users

aqua-src/test-import.aqua

@@ -1,6 +0,0 @@
-import "builtin.aqua"
-
-func foo() -> u64:
-    t <- Peer.timestamp_sec()
-    id()
-    <- t

model/src/main/scala/aqua/model/ScriptModel.scala

@@ -4,35 +4,28 @@ import aqua.model.func.{FuncCallable, FuncModel}
 import cats.Monoid
 import cats.data.Chain
 
-// TODO make one chain to have order
 case class ScriptModel(
   models: Chain[Model] = Chain.empty
 ) extends Model {
 
   case class Acc(
-    arrows: Map[String, FuncCallable],
-    values: Map[String, ValueModel]
+    arrows: Map[String, FuncCallable] = Map.empty,
+    values: Map[String, ValueModel] = Map.empty,
+    output: Chain[FuncCallable] = Chain.empty
   )
 
   lazy val funcs: Chain[FuncModel] = models.collect { case c: FuncModel => c }
   lazy val constants: Chain[ConstantModel] = models.collect { case c: ConstantModel => c }
 
-  def resolveFunctions: Chain[FuncCallable] = {
-    val constantsToName =
-      constants.map(c => c.name -> c.value).toList.toMap
-
-    funcs
-      .foldLeft(
-        (
-          Map.empty[String, FuncCallable],
-          Chain.empty[FuncCallable]
-        )
-      ) { case ((acc, outputAcc), func) =>
-        val fr = func.capture(acc, constantsToName)
-        acc.updated(func.name, fr) -> outputAcc.append(fr)
-      }
-      ._2
-  }
+  lazy val resolveFunctions: Chain[FuncCallable] = models
+    .foldLeft(Acc()) {
+      case (a, c: ConstantModel) => a.copy(values = a.values.updated(c.name, c.value))
+      case (a, func: FuncModel) =>
+        val fr = func.capture(a.arrows, a.values)
+        a.copy(output = a.output :+ fr, arrows = a.arrows.updated(func.name, fr))
+      case (a, _) => a
+    }
+    .output
 }
 
 object ScriptModel {
@@ -47,11 +40,11 @@ object ScriptModel {
   }
 
   // Builds a ScriptModel if given model can be considered as a part of a script
-  def toScriptPart(m: Model): Option[ScriptModel] = m match {
-    case fm: FuncModel => Some(ScriptModel(models = Chain.one(fm)))
-    case sm: ServiceModel => Some(ScriptModel(models = Chain.one(sm)))
-    case tm: TypeModel => Some(ScriptModel(models = Chain.one(tm)))
-    case cm: ConstantModel => Some(ScriptModel(models = Chain.one(cm)))
-    case _ => None
-  }
+  def toScriptPart(m: Model): Option[ScriptModel] = Option(m).filter {
+    case _: FuncModel => true
+    case _: ServiceModel => true
+    case _: TypeModel => true
+    case _: ConstantModel => true
+    case _ => false
+  }.map(Chain.one).map(ScriptModel(_))
 }

model/src/main/scala/aqua/model/transform/Topology.scala

@@ -1,7 +1,7 @@
 package aqua.model.transform
 
 import aqua.model.ValueModel
-import aqua.model.func.body.{CallServiceTag, FuncOps, OnTag, OpTag, SeqTag}
+import aqua.model.func.body.{CallServiceTag, FuncOps, OnTag, OpTag, ParTag, SeqTag}
 import cats.Eval
 import cats.data.Chain
 import cats.free.Cofree
@@ -9,6 +9,9 @@ import cats.free.Cofree
 object Topology {
   type Tree = Cofree[Chain, OpTag]
 
+  def rightBoundary(root: Tree): List[OpTag] =
+    root.head :: root.tailForced.lastOption.fold(List.empty[OpTag])(rightBoundary)
+
   // Walks through peer IDs, doing a noop function on each
   // If same IDs are found in a row, does noop only once
   // TODO: if there's a chain like a -> b -> c -> ... -> b -> g, remove everything between b and b
@@ -61,25 +64,24 @@ object Topology {
 
         def modifyChildrenList(list: List[Tree], prev: Option[Tree]): Chain[Tree] = list match {
           case Nil => Chain.empty
-          case op :: Nil =>
-            prev match {
-              // TODO: sequence might be longer
-              case Some(Cofree(SeqTag, seqTail)) =>
-                seqTail.value.lastOption match {
-                  case Some(Cofree(ont: OnTag, _)) =>
-                    through(ont.via.reverse ++ pathViaChain) :+ op
-                  case _ =>
-                    Chain.one(op)
-                }
-              case _ =>
-                Chain.one(op)
-            }
+          case op :: tail =>
+            // TODO further improve
+            val prevPath = Chain
+              .fromSeq(prev.toList.flatMap(rightBoundary).takeWhile {
+                case ParTag => false
+                case _ => true
+              })
+              .collect { case OnTag(_, v) =>
+                v.reverse
+              }
+
+            if (prevPath.isEmpty) op +: modifyChildrenList(tail, Some(op))
+            else
+              through(prevPath.flatMap(identity) ++ pathViaChain).append(op) ++ modifyChildrenList(
+                tail,
+                Some(op)
+              )
 
-          case (oncf @ Cofree(ont: OnTag, _)) :: op :: tail =>
-            (oncf +: through(ont.via.reverse ++ pathViaChain)) ++ modifyChildrenList(
-              op :: tail,
-              Some(oncf)
-            )
           case o :: ops => o +: modifyChildrenList(ops, Some(o))
         }
 

model/src/test/scala/aqua/model/Node.scala

@@ -1,7 +1,16 @@
 package aqua.model
 
 import aqua.model.func.Call
-import aqua.model.func.body.{CallServiceTag, FuncOp, FuncOps, OnTag, OpTag, SeqTag, XorTag}
+import aqua.model.func.body.{
+  CallServiceTag,
+  FuncOp,
+  FuncOps,
+  MatchMismatchTag,
+  OnTag,
+  OpTag,
+  SeqTag,
+  XorTag
+}
 import aqua.model.transform.BodyConfig
 import aqua.types.ScalarType
 import cats.Eval
@@ -134,6 +143,12 @@ object Node {
       body.toList
     )
 
+  def _match(l: ValueModel, r: ValueModel, body: Node) =
+    Node(
+      MatchMismatchTag(l, r, shouldMatch = true),
+      body :: Nil
+    )
+
   def through(peer: ValueModel): Node =
     FuncOps.noop(peer).tree
 }

model/src/test/scala/aqua/model/transform/TopologySpec.scala

@@ -136,6 +136,78 @@ class TopologySpec extends AnyFlatSpec with Matchers {
       )
     )
 
+//    println(Console.BLUE + init)
+//    println(Console.YELLOW + proc)
+//    println(Console.MAGENTA + expected)
+//    println(Console.RESET)
+
+    proc.equalsOrPrintDiff(expected) should be(true)
+  }
+
+  "topology resolver" should "get back to init peer after a long chain" in {
+
+    val init = on(
+      initPeer,
+      relay :: Nil,
+      seq(
+        on(
+          otherPeer,
+          otherRelay :: Nil,
+          call(0),
+          on(
+            otherPeer2,
+            otherRelay :: Nil,
+            call(1),
+            _match(
+              otherPeer,
+              otherRelay,
+              on(
+                otherPeer,
+                otherRelay :: Nil,
+                call(2)
+              )
+            )
+          )
+        ),
+        call(3)
+      )
+    )
+
+    val proc: Node = Topology.resolve(init)
+
+    val expected = on(
+      initPeer,
+      relay :: Nil,
+      seq(
+        on(
+          otherPeer,
+          otherRelay :: Nil,
+          through(relay),
+          through(otherRelay),
+          call(0, otherPeer),
+          on(
+            otherPeer2,
+            otherRelay :: Nil,
+            through(otherRelay),
+            call(1, otherPeer2),
+            _match(
+              otherPeer,
+              otherRelay,
+              on(
+                otherPeer,
+                otherRelay :: Nil,
+                through(otherRelay),
+                call(2, otherPeer)
+              )
+            )
+          )
+        ),
+        through(otherRelay),
+        through(relay),
+        call(3, initPeer)
+      )
+    )
+
 //    println(Console.BLUE + init)
 //    println(Console.YELLOW + proc)
 //    println(Console.MAGENTA + expected)