adopted to 2.12

Author: rssh

build.sbt

@@ -3,22 +3,26 @@ name:="scala-gopher"
 
 organization:="com.github.rssh"
 
-scalaVersion := "2.11.8"
+//scalaVersion := "2.12.0"
+scalaVersion := "2.12.1-SNAPSHOT"
 
 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.7-SNAPSHOT-p1"
 //libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.6-SNAPSHOT"
 
-libraryDependencies += "org.scalatest" %% "scalatest" % "2.2.6" % "test"
+libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.0" % "test"
 
-libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.4.8"
+libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.4.12"
 
 //TODO: enable after 1.0
 //libraryDependencies += "com.typesafe.akka" %% "akka-stream-experimental" % "0.9"

notes/techreport.pdf

@@ -7,6 +7,7 @@
 }
 \date{\today}
 
+% arxiv url: https://arxiv.org/abs/1611.00602
 
 
 \documentclass[12pt]{article}

notes/techreport.tex

@@ -5,11 +5,12 @@ import akka.actor._
 import akka.pattern._
 import scala.concurrent._
 import scala.concurrent.duration._
-import gopher._
+import scala.util._
 import scala.language.experimental.macros
 import scala.language.postfixOps
 import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
+import gopher._
 
 class ActorBackedChannel[A](futureChannelRef: Future[ActorRef], override val api: GopherAPI) extends Channel[A]
 {
@@ -26,18 +27,21 @@ class ActorBackedChannel[A](futureChannelRef: Future[ActorRef], override val api
                                }
                       }
    }
+   implicit val ec = api.executionContext
    if (closed) {
      if (closedEmpty) {
        applyClosed();
      } else {
          // TODO: ask timeput on closed channel set in config.
          futureChannelRef.foreach{ ref => val f = ref.ask(ClosedChannelRead(cont))(5 seconds)
-                                     f.onFailure{
-                                          case e: AskTimeoutException => applyClosed()
-                                     }
-                                     f.onSuccess{
-                                          case ChannelCloseProcessed(0) =>
-                                                                  closedEmpty = true
+                                     f.onComplete{
+                                         case Failure(e) =>
+                                               if (e.isInstanceOf[AskTimeoutException]) {
+                                                 applyClosed()
+                                               }
+                                         case Success(ChannelCloseProcessed(0)) =>
+                                               closedEmpty = true
+                                         case _ =>  // do nothing
                                      }
                                  }
      }
@@ -47,33 +51,33 @@ class ActorBackedChannel[A](futureChannelRef: Future[ActorRef], override val api
   }
 
   private def  contRead[B](x:ContRead[A,B]): Unit =
-     futureChannelRef.foreach( _ ! x )
+     futureChannelRef.foreach( _ ! x )(api.executionContext)
 
   def  cbwrite[B](f: ContWrite[A,B] => Option[(A,Future[Continuated[B]])], flwt: FlowTermination[B] ): Unit = 
     if (closed) {
       flwt.doThrow(new ChannelClosedException())
     } else {
-     futureChannelRef.foreach( _ ! ContWrite(f,this, flwt) )
+     futureChannelRef.foreach( _ ! ContWrite(f,this, flwt) )(api.executionContext)
     }
 
   private def contWrite[B](x:ContWrite[A,B]): Unit =
-    futureChannelRef.foreach( _ ! x )
+    futureChannelRef.foreach( _ ! x )(api.executionContext)
 
-  private[this] implicit val ec = api.executionContext
+  //private[this] implicit val ec = api.executionContext
 
   def isClosed: Boolean = closed
 
   def close(): Unit =
   {
-    futureChannelRef.foreach( _ ! ChannelClose )
+    futureChannelRef.foreach( _ ! ChannelClose )(api.executionContext)
     closed=true
   }
 
 
   override protected def finalize(): Unit =
   {
    // allow channel actor be grabage collected
-   futureChannelRef.foreach( _ ! ChannelRefDecrement )
+   futureChannelRef.foreach( _ ! ChannelRefDecrement )(api.executionContext)
   }
 
   private var closed = false

src/main/scala/gopher/channels/ActorBackedChannel.scala

@@ -540,9 +540,13 @@ class FoldSelectorBuilderImpl(override val c:Context) extends SelectorBuilderImp
                             c.abort(tp.pos, "match pattern in select without guard must be in form x:channel.write or x:channel.read");
                          } else {
                           parseGuardInSelectorCaseDef(termName, caseDef.guard) match {
-                               case q"scala.async.Async.await[${t}](${readed}.aread):${t1}" =>
+                            case q"scala.async.Async.await[${t}](${readed}.aread):${t1}" =>
                                         onRead(readed)
-                               case q"scala.async.Async.await[${t}](${ch}.awrite($expression)):${t1}" =>
+                            case q"gopher.goasync.AsyncWrapper.await[${t}](${readed}.aread):${t1}" =>
+                                        onRead(readed)
+                            case q"scala.async.Async.await[${t}](${ch}.awrite($expression)):${t1}" =>
+                                        onWrite(ch)
+                            case q"gopher.goasync.AsyncWrapper.await[${t}](${ch}.awrite($expression)):${t1}" =>
                                         onWrite(ch)
                                case x@_ =>
                                   c.abort(tp.pos, "can't parse match guard: "+x);

src/main/scala/gopher/channels/FoldSelectorBuilder.scala

@@ -101,7 +101,10 @@ class SelectorBuilderImpl(val c: Context) extends ASTUtilImpl
             c.abort(f.tree.pos, "match expected in gopher select loop, have: ${MacroUtil.shortString(f.tree)}");
        }
     }
-    c.Expr[T](c.untypecheck(q"scala.async.Async.await(${builder}.go)"))
+    System.err.println(s"builder=${builder}")
+    //val tc = c.typecheck(builder)
+    //System.err.println(s"tc=${tc}")
+    c.Expr[T](MacroUtil.cleanUntypecheck(c)(q"gopher.goasync.AsyncWrapper.await(${builder}.go)"))
    }
 
    def foreachBuildMatch(cases:List[c.universe.CaseDef]):c.Tree =
@@ -330,7 +333,7 @@ class SelectorBuilderImpl(val c: Context) extends ASTUtilImpl
             c.abort(f.tree.pos, "match expected in gopher select map, have: ${MacroUtil.shortString(f.tree)}");
 
      }
-     c.Expr[Input[T]](c.untypecheck(q"${builder}.started"))
+     c.Expr[Input[T]](MacroUtil.cleanUntypecheck(c)(q"${builder}.started"))
    }
 
    def builder[T](f:c.Expr[PartialFunction[Any,T]]):c.Tree =
@@ -364,7 +367,7 @@ class SelectorBuilderImpl(val c: Context) extends ASTUtilImpl
                          mapBuildMatch[T](cases)
         case _ => c.abort(f.tree.pos,"expected partial function with syntax case ... =>, have ${MacroUtil.shortString(f.tree)}");
      }
-     c.Expr[Input[T]](c.untypecheck(q"${builder}.started"))
+     c.Expr[Input[T]](MacroUtil.cleanUntypecheck(c)(q"${builder}.started"))
    }
 
 }
@@ -455,16 +458,16 @@ object SelectorBuilder
      val ftParam = ValDef(Modifiers(Flag.PARAM),ft,tq"gopher.FlowTermination[${weakTypeOf[T]}]",EmptyTree)
      val ecParam = ValDef(Modifiers(Flag.PARAM),ec,tq"scala.concurrent.ExecutionContext",EmptyTree)
      val nvaldefs = ecParam::ftParam::valdefs
-     val asyncBody = GoAsync.transformAsyncBody[T](c)(body)
+     val asyncBody =  GoAsync.transformAsyncBody[T](c)(body)
      val nbody = q"""{
                       implicit val ${ft1} = ${ft}
                       implicit val ${ec1} = ${ec}
-                      scala.async.Async.async(${transformDelayedMacroses[T](c)(asyncBody)})(${ec})
+                      gopher.goasync.AsyncWrapper.async(${transformDelayedMacroses[T](c)(asyncBody)})(${ec})
                      }
                   """
      val newTree = lastFun(nvaldefs,nbody)
      // untypecheck is necessory: otherwise exception in async internals
-     c.Expr[S](c.untypecheck(newTree))
+     c.Expr[S](MacroUtil.cleanUntypecheck(c)(newTree))
    }
 
    def idleImpl[T:c.WeakTypeTag,S](c:Context)(body:c.Expr[T]):c.Expr[S] =

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

@@ -198,27 +198,42 @@ object GoAsync
    {
     import c.universe._
     val btype = body.tpe
-    // untypechack is necessory, because async-transform later will corrupt
-    //  symbols owners inside body
-    // [scala-2.11.8]
-    val nb = c.untypecheck(body)
+    val nb = body 
     val anb = atPos(body.pos){
        val nnb = transformAsyncBody(c)(nb)
        val ec = c.inferImplicitValue(c.weakTypeOf[ExecutionContext])
-       q"(${param})=>scala.async.Async.async[$btype](${nnb})($ec)"
+       // untypechack is necessory, because async-transform corrupt
+       //  symbols owners inside body and we change scope of param
+       // [scala-2.11.8]
+       MacroUtil.cleanUntypecheck(c)(
+          q"(${param})=>scala.async.Async.async[$btype](${nnb})($ec)"
+       )
     }
     val ar = atPos(fun.pos) {
        val uar = if (implicitParams.isEmpty) {
           q"gopher.asyncApply1(${fun})(${anb})"
        } else {
           q"gopher.goasync.AsyncApply.apply1i(${fun})(${anb},${implicitParams})"
        }
-       // 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(uar)
+       if (true) {
+         // 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
+         try {
+          c.typecheck(uar)
+         }catch{
+          case ex: Exception =>
+            System.err.println(s"failed to typecheck uar: ${ex}")
+            System.err.println(s"uar=${uar}")
+            System.err.println(s"nb=${nb}")
+            //System.err.println(s"raw=${showRaw(uar)}")
+            throw ex
+         }
+       } else {
+         uar
+       }
     }
     //typecheck with macros disabled is needed for compiler,
     //to set symbol 'await', because async macro discovered

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

@@ -1,5 +1,6 @@
 package gopher.util
 
+import scala.annotation._
 import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
 import scala.language.reflectiveCalls
@@ -53,6 +54,63 @@ object MacroUtil
    findAwait.found
   }
 
+  /**
+   * bug in async/scala-2.12.x
+   * async/types generate in state-machine next chunk of code:
+   *```
+   *    val result: scala.concurrent.Promise[Int] = Promise.apply[Int]();
+   *    <stable> <accessor> def result: scala.concurrent.Promise[Int] = stateMachine$macro$1041.this.result;
+   *    val execContext: scala.concurrent.ExecutionContext = ..
+   *    <stable> <accessor> def execContext: scala.concurrent.Promise[Int] = stateMachine$macro$1041.this.execContext;
+   *```
+   *  when we attempt untype/type code again, it is not compiled.
+   *So, we need to remove result and execContext DefDefs
+   **/
+  def removeAsyncStateMachineResultDefDef(c:Context)(tree: c.Tree):c.Tree =
+  {
+   import c.universe._
+
+   val outsideStm  = new Transformer {
+
+      override def transform(tree:Tree):Tree =
+         tree match {
+           case ClassDef(mods,name,tparams,impl) 
+                   if (name.toString.startsWith("stateMachine$")) =>
+                      impl match {
+                        case Template(parents,self,body) =>
+                               ClassDef(mods,name,tparams,
+                                           Template(parents,self,removeResultDefDef(body,Nil)))
+                        //case _ => // impossible, throw
+                      }
+           case _ => super.transform(tree)
+         }
+
+      @tailrec
+      def removeResultDefDef(body:List[Tree],acc:List[Tree]):List[Tree] =
+      {
+        body match {
+          case Nil => acc.reverse
+          case head::tail =>
+                  val (rest,nacc) = head match {
+                    case DefDef(mods,name,tparams,vparamss,tpt,rsh) 
+                                          if (name.toString == "result" ||
+                                              name.toString == "execContext" ) => (tail,acc)
+                    case _ => (tail, transform(head)::acc)
+                  }
+                  removeResultDefDef(rest,nacc)
+        }
+      }
+
+   }
+
+   val retval = outsideStm.transform(tree)
+   retval
+  }
+
+  def cleanUntypecheck(c:Context)(tree:c.Tree):c.Tree =
+  {
+    removeAsyncStateMachineResultDefDef(c)(c.untypecheck(tree))
+  }
 
 
   final val SHORT_LEN = 80

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

@@ -9,9 +9,21 @@ object ReflectUtil
 
    def retrieveValSymbols[T:u.TypeTag](u:Universe)(ownerType:u.Type): List[u.TermSymbol] =
    {
-     ownerType.members.filter(_.isTerm).map(_.asTerm).filter( x =>
-                                          x.isVal && x.typeSignature <:< u.typeOf[T]
-     ).toList
+     val r1 = ownerType.members.filter(_.isTerm).map(_.asTerm).filter(x => x.isVal)
+     val signatures = r1.map(_.typeSignature)
+     val ut = u.weakTypeOf[T]
+     val checkResults = signatures.map( _ <:< u.typeOf[T])
+     val retval = ownerType.members.filter(_.isTerm).map(_.asTerm).filter{ x =>
+                         if (x.isVal) {
+                            // in scala 2.12 getter method type, scala 2.11 - type
+                            val r = x.typeSignature match {
+                              case u.NullaryMethodType(rt) => rt <:< u.typeOf[T] // for scala-2.12
+                              case _ => (x.typeSignature <:< u.typeOf[T])   // for scala-2.11
+                            }
+                            r
+                         } else false
+     }.toList
+     retval
    }
 
 

src/main/scala/gopher/util/ReflectUtil.scala

@@ -77,7 +77,7 @@ object Sieve
 class SieveSuite extends FunSuite
 {
 
- test("last prime before 1000", Now) {
+ test("last prime before 1000") {
 
    val quit = Promise[Boolean]()
    val quitInput = futureInput(quit.future)