implemented handling async operations inside foreach.

Author: rssh

build.sbt

@@ -9,11 +9,12 @@ resolvers += Resolver.sonatypeRepo("snapshots")
 
 resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/"
 
-scalacOptions ++= Seq("-unchecked","-deprecation", "-feature" /* , "-Ymacro-debug-lite" ,  "-Ydebug"  , "-Ylog:lambdalift" */ )
+scalacOptions ++= Seq("-unchecked","-deprecation", "-feature" /* ,  "-Ymacro-debug-lite"  ,  "-Ydebug"  ,  "-Ylog:lambdalift"  */ )
 
 libraryDependencies <+= scalaVersion( "org.scala-lang" % "scala-reflect" % _ )
 
-libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.6-RC2"
+//libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.6-RC2"
+libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.6-SNAPSHOT"
 
 libraryDependencies += "org.scalatest" %% "scalatest" % "2.2.6" % "test"
 

src/main/scala/gopher/channels/SelectorBuilder.scala

@@ -279,7 +279,7 @@ object SelectorBuilder
 
     val symbolsToErase = Set(caseDef.pat.symbol, caseDef.pat.symbol.owner)
 
-    // Loook's like bug in 'untypecheck' : when we split cassDef on few functions, than sometines, symbols
+    //  when we split cassDef on few functions, than sometines, symbols
     // entries in identifier tree are not cleared.  
     //   So, we 'reset' symbols which belong to caseDef which will be erased by macros
     //   //TODO: check, may be will be better to use scala-compiler internal API and changeOwner instead.

src/main/scala/gopher/goasync/AsyncApply.scala

@@ -0,0 +1,71 @@
+package gopher.goasync
+
+import scala.language.experimental.macros
+import scala.language.reflectiveCalls
+import scala.reflect.macros.blackbox.Context
+import scala.reflect.api._
+import gopher._
+import gopher.util._
+import scala.concurrent._
+import scala.annotation.unchecked._
+
+
+
+object AsyncApply
+{
+
+
+  def impl1[A:c.WeakTypeTag,B:c.WeakTypeTag,C:c.WeakTypeTag](c:Context)(hof:c.Expr[Function[Function[A,B],C]])(nf:c.Expr[Function[A,Future[B]]]):
+                                                                                    c.Expr[Future[C]] =
+  {
+    import c.universe._
+    val nhof = transformHof[A,B,C](c)(hof.tree)
+    val retval = c.Expr[Future[C]](q"${nhof}(${nf})")
+    System.err.println("retval:"+retval)
+    retval
+  }
+
+  def transformHof[A:c.WeakTypeTag,B:c.WeakTypeTag,C:c.WeakTypeTag](c:Context)(hof:c.Tree):c.Tree = {
+    import c.universe.{Function=>_,_}
+    hof match {
+      case q"${p}.$h" =>
+               val ah = genAsyncName(c)(h,hof.pos)
+               q"${p}.${ah}"
+      case q"${p}.$h[$w]" =>
+               val ah = genAsyncName(c)(h,hof.pos)
+               q"${p}.${ah}[$w]"
+      case q"($fp)=>$res($fp1)" if (fp.symbol == fp1.symbol) => 
+               val nested = transformHof[A,B,C](c)(res)
+               val nname = TermName(c.freshName())
+               val paramType = tq"Function[${c.weakTypeOf[A]},Future[${c.weakTypeOf[B]}]]"
+               //val paramDef = q"val $nname:${paramType}"
+               val paramDef = ValDef(Modifiers(Flag.PARAM),nname,paramType,EmptyTree)
+               c.typecheck(q"($paramDef)=>$nested($nname)")
+      case q"{ ..$stats }" =>
+                val nstats = transformLast(c){
+                               t => transformHof[A,B,C](c)(t)
+                             }(stats)
+           q"{ ..$nstats }"
+      case _ => c.abort(hof.pos,"hof match failed:"+hof)
+    }
+  }
+
+  def genAsyncName(c:Context)(h:c.TermName,pos:c.Position):c.TermName =
+  {
+    import c.universe._
+    h match {
+       case TermName(hname) => 
+              TermName(hname+"Async")
+       case _ =>
+              c.abort(pos,"ident expected for hight order function")
+    }
+  }
+
+  def transformLast(c:Context)(f:c.Tree=>c.Tree)(block: List[c.Tree]):List[c.Tree] =
+   block match {
+     case Nil => Nil
+     case r::Nil => f(r)::Nil
+     case h::q => h::transformLast(c)(f)(q)
+   }
+
+}

src/main/scala/gopher/goasync/AsyncIterable.scala

@@ -0,0 +1,32 @@
+package gopher.goasync
+
+import scala.concurrent._
+import scala.async.Async._
+import scala.collection.generic._
+
+
+class AsyncIterable[T](val x:Iterable[T]) //extends AnyVal [implementation restriction, [scala-2.11.8]
+{
+
+   def foreachAsync[U](f: T => Future[U])(implicit ec:ExecutionContext): Future[Unit] =
+   async{
+     val it = x.iterator
+     while(it.hasNext) {
+        await(f(it.next))
+     }
+   }
+
+   def mapAsync[U,That](f: T => Future[U])(implicit bf: CanBuildFrom[Iterable[T],U,That], ec: ExecutionContext): Future[That] =
+   {
+     async {
+      val builder = bf.apply()
+      val it = x.iterator
+      while(it.hasNext) {
+         builder += await(f(it.next))
+      }
+      builder.result()
+     }
+   }
+
+}
+

src/main/scala/gopher/goasync/GoAsync.scala

@@ -13,7 +13,26 @@ import scala.annotation.unchecked._
 /**
  * async arround go. 
  *
- * Basicly go is wrapped inside SIP-22 async with defer
+ * Basicly go is 
+ *   1. translate await-like exressions inside inline functions to calls of appropriative async functions.
+ *      (or show error if not found).
+ *```
+ *      x.foreach{ x => p; await(x); .. }
+ *```
+ *  become
+ *```
+ *      await( transform-defer( x.foreachAsync{ x => async(p; await(x); ..) }) )
+ *```
+ *    (note, that channel.read macroses are expanded to await-s on this point)
+ *
+ *   2. transform defer calls if defer statement is found inside go:
+ *```
+ *   asnyc{ p .. defer(x) ..  }
+ *```
+ *   become (reallity is a little complext, here is just idea)
+ *```
+ *   { val d = new Defers(); async{  p .. d.defer(x) ..  }.onComplete(d.tryProcess) }
+ *```
  */
 object GoAsync
 {
@@ -23,26 +42,27 @@ object GoAsync
    def goImpl[T:c.WeakTypeTag](c:Context)(body:c.Expr[T])(ec:c.Expr[ExecutionContext]):c.Expr[Future[T]] =
    {
      import c.universe._
-     if (containsDefer(c)(body)) {
+     val nbody = GoAsync.transformAsyncBody[T](c)(body.tree)
+     val r = if (containsDefer(c)(body)) {
        val defers = TermName(c.freshName)
        val promise = TermName(c.freshName)
        // asyn transform wantstyped tree on entry, so we must substitute 'defers' to untyped 
        // values after it, no before.
-       c.Expr[Future[T]](
                          q"""
                              gopher.goasync.GoAsync.transformDeferMacro[${c.weakTypeOf[T]}](
                                {implicit val ${defers} = new Defers[${c.weakTypeOf[T]}]()
                                 val ${promise} = Promise[${c.weakTypeOf[T]}]()
-                                scala.async.Async.async(${body})(${ec}).onComplete( x =>
+                                scala.async.Async.async(${nbody})(${ec}).onComplete( x =>
                                      ${promise}.complete(${defers}.tryProcess(x))
                                 )(${ec})
                                 ${promise}.future
                                }
                              )
-                          """)
+                          """
      } else {
-       c.Expr[Future[T]](q"scala.async.Async.async(${body})(${ec})")
+       q"scala.async.Async.async(${nbody})(${ec})"
      }
+     c.Expr[Future[T]](r)
    }
 
    def goScopeImpl[T:c.WeakTypeTag](c:Context)(body:c.Expr[T]):c.Expr[T] =
@@ -101,6 +121,86 @@ object GoAsync
     transformer.transform(body)
    }
 
+   def transformAsyncBody[T:c.WeakTypeTag](c:Context)(body:c.Tree):c.Tree = 
+   {
+    import c.universe._
+    val transformer = new Transformer {
+      override def transform(tree:Tree):Tree =
+       tree match {
+         case q"${f1}(${a}=>${b})" =>
+            val found = findAwait(c)(b)
+            if (found) {
+                //System.err.println(s"found hof entry ${f1} ${a}=>${b}")
+                val btype = b.tpe
+                // untypechack is necessory, because async-transform later will corrupt
+                //  symbols owners inside b
+                // [scala-2.11.8]
+                val nb = c.untypecheck(b)
+                val anb = atPos(b.pos){
+                              // typecheck here is needed to prevent incorrect liftingUp of
+                              // internal variables in ${b}
+                              //[scala-2.11.8]
+                              //c.typecheck(q"(${a})=>go[${btype}](${nb})")
+                              val nnb = transformAsyncBody(c)(nb)
+                              //c.typecheck(q"(${a})=>scala.async.Async.async[${btype}](${nnb})")
+                              q"(${a})=>scala.async.Async.async[${btype}](${nnb})"
+                          }
+                val ar = atPos(tree.pos){
+                          // typecheck is necessory
+                          //  1. to prevent runnint analysis of async over internal awaits in anb as on 
+                          //    enclosing async instead those applied from asyncApply
+                          //  2. to expand macroses here, to prevent error during expanding macroses
+                          //    in next typecheck
+                          c.typecheck(q"gopher.asyncApply1(${f1})(${anb})")
+                          //q"gopher.asyncApply1(${f1})(${anb})"
+                        }
+                //typecheck with macros disabled is needed for compiler,
+                //to set symbol 'await', because async macro discovered
+                //awaits by looking at symbole
+                val r = c.typecheck(q"scala.async.Async.await[${btype}](${ar})",withMacrosDisabled=true)
+                r
+            } else {
+                super.transform(tree)
+            }
+         case _ =>
+                super.transform(tree)
+       }
+    }
+    transformer.transform(body)
+   }
 
+   def findAwait(c:Context)(body:c.Tree): Boolean = 
+   {
+    import c.universe._
+    var found: Boolean = false
+    val transformer = new Transformer {
+
+      override def transform(tree:Tree):Tree =
+      {
+         if (found) 
+          tree
+         else {
+           tree match {
+             case q"(scala.async.Async.await[${w}]($r)):${w1}"=>
+                     System.err.println(s"found await: [${w}](${r})")
+                     found = true
+                     // here we erase 'await' symbols
+                     //q"(scala.async.Async.await[${w}]($r)):${w1}"
+                     tree
+             case q"(${a}=>${b})" =>
+                     // don't touch nested functions
+                     tree
+             case _ => 
+                  super.transform(tree)
+            }
+         }
+      }
+
+    }
+    transformer.transform(body)
+    found
+   }
+
+
 }
 

src/main/scala/gopher/package.scala

@@ -2,6 +2,10 @@
 import scala.language.experimental.macros
 import scala.language.implicitConversions
 
+import scala.concurrent._
+import gopher.channels._
+import gopher.goasync._
+
 /**
  * Provides scala API for 'go-like' CSP channels.
  * 
@@ -66,9 +70,6 @@ import scala.language.implicitConversions
 package object gopher {
 
 
-import scala.concurrent._
-import gopher.channels._
-import gopher.goasync._
 
  //
  // magnetic arguments for selector-builder unsugared API
@@ -172,7 +173,7 @@ import gopher.goasync._
    def aread: Future[T] = f
  }
 
- //import scala.language.experimental.macros
+ import scala.language.experimental.macros
  import scala.reflect.macros.blackbox.Context
  import scala.reflect.api._
  def awaitImpl[T](c:Context)(v:c.Expr[Future[T]]):c.Expr[T] =
@@ -181,5 +182,11 @@ import gopher.goasync._
    c.Expr[T](q"scala.async.Async.await($v)")
  }
 
+
+ def asyncApply1[A,B,C](hof:(A=>B)=>C)(nf:A=>Future[B]):Future[C] =
+                          macro gopher.goasync.AsyncApply.impl1[A,B,C]
+
+ implicit def toAsyncIterable[T](x:Iterable[T]):AsyncIterable[T] = new AsyncIterable[T](x)
+
 }
 

src/main/scala/gopher/util/MacroUtil.scala

@@ -54,5 +54,6 @@ object MacroUtil
   }
 
 
+
   final val SHORT_LEN = 80
 }

src/test/scala/gopher/hofasyn/FibbonacyAsyncLoopSuite.scala

@@ -0,0 +1,66 @@
+package gopher.hofasyn
+
+import gopher._
+import gopher.channels._
+import scala.language._
+import scala.concurrent._
+import scala.concurrent.duration._
+import scala.async.Async._
+
+import org.scalatest._
+import gopher.tags._
+
+
+/*
+* code from go tutorial: http://tour.golang.org/#66
+* 
+*/
+object FibonaccyL {
+
+  import scala.concurrent.ExecutionContext.Implicits.global
+  
+  def fibonacci(c: Output[Long], quit: Input[Int]): Future[Unit] =
+  go {
+     var (x,y) = (0L,1L)
+     for(s <- gopherApi.select.forever) {
+      s match {
+        case z: c.write if (z == x) => 
+                      x = y 
+                      y = z+y
+        case q: quit.read =>
+                 implicitly[FlowTermination[Unit]].doExit(())
+      }
+     }
+  }
+  
+  def run(n:Int, acceptor: Long => Unit ): Future[Unit] =
+  {
+    val c = gopherApi.makeChannel[Long](1);
+    val quit = gopherApi.makeChannel[Int](1);
+    go {
+      for(i <-1 to n) {
+         val xLLFind = c.read
+         //Console.println(s"received: ${i}, ${xLLFind}")
+         acceptor(xLLFind)
+      }
+      //System.err.println("sending quit")
+      quit <~ 0
+    }
+
+    fibonacci(c,quit)
+  }
+  
+  def gopherApi = CommonTestObjects.gopherApi 
+  
+}
+
+class FibonaccyAsyncLoopSuite extends FunSuite
+{
+  
+  test("fibonaccy must be processed up to 50") {
+    @volatile var last:Long = 0;
+    Await.ready( FibonaccyL.run(50, last = _ ), 10 seconds )
+    assert(last != 0)
+  }
+
+}