src/main/scala/scala/async/continuations/AsyncBaseWithCPSFallback.scala


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

/*
 * Copyright (C) 2012 Typesafe Inc. <http://www.typesafe.com>
 */

package scala.async
package continuations

import scala.language.experimental.macros

import scala.reflect.macros.Context
import scala.util.continuations._

trait AsyncBaseWithCPSFallback extends AsyncBase {

  /* Fall-back for `await` using CPS plugin.
   * 
   * Note: This method is public, but is intended only for internal use.
   */
  def awaitFallback[T](awaitable: futureSystem.Fut[T]): T @cps[futureSystem.Fut[Any]]

  override protected[async] def fallbackEnabled = true

  /* Implements `async { ... }` using the CPS plugin.
   */
  protected def cpsBasedAsyncImpl[T: c.WeakTypeTag](c: Context)(body: c.Expr[T]): c.Expr[futureSystem.Fut[T]] = {
    import c.universe._

    def lookupMember(name: String) = {
      val asyncTrait = c.mirror.staticClass("scala.async.continuations.AsyncBaseWithCPSFallback")
      val tpe = asyncTrait.asType.toType
      tpe.member(newTermName(name)).ensuring(_ != NoSymbol)
    }

    AsyncUtils.vprintln("AsyncBaseWithCPSFallback.cpsBasedAsyncImpl")

    val utils            = TransformUtils[c.type](c)
    val futureSystemOps  = futureSystem.mkOps(c)
    val awaitSym         = utils.defn.Async_await
    val awaitFallbackSym = lookupMember("awaitFallback")

    // replace `await` invocations with `awaitFallback` invocations
    val awaitReplacer = new Transformer {
      override def transform(tree: Tree): Tree = tree match {
        case Apply(fun @ TypeApply(_, List(futArgTpt)), args) if fun.symbol == awaitSym =>
          val typeApp = treeCopy.TypeApply(fun, Ident(awaitFallbackSym), List(TypeTree(futArgTpt.tpe)))
          treeCopy.Apply(tree, typeApp, args.map(arg => c.resetAllAttrs(arg.duplicate)))
        case _ =>
          super.transform(tree)
      }
    }
    val bodyWithAwaitFallback = awaitReplacer.transform(body.tree)

    /* generate an expression that looks like this:
         reset {
           val f = future { ... }
           ...
           val x = awaitFallback(f)
           ...
           future { expr }
         }.asInstanceOf[Future[T]]
     */

    val bodyWithFuture = {
      val tree = bodyWithAwaitFallback match {
        case Block(stmts, expr) => Block(stmts, futureSystemOps.spawn(expr))
        case expr               => futureSystemOps.spawn(expr)
      }
      c.Expr[futureSystem.Fut[Any]](c.resetAllAttrs(tree.duplicate))
    }

    val bodyWithReset: c.Expr[futureSystem.Fut[Any]] = reify {
      reset { bodyWithFuture.splice }
    }
    val bodyWithCast = futureSystemOps.castTo[T](bodyWithReset)

    AsyncUtils.vprintln(s"CPS-based async transform expands to:\n${bodyWithCast.tree}")
    bodyWithCast
  }

  override def asyncImpl[T: c.WeakTypeTag](c: Context)(body: c.Expr[T]): c.Expr[futureSystem.Fut[T]] = {
    AsyncUtils.vprintln("AsyncBaseWithCPSFallback.asyncImpl")

    val analyzer = AsyncAnalysis[c.type](c, this)

    if (!analyzer.reportUnsupportedAwaits(body.tree))
      super.asyncImpl[T](c)(body)   // no unsupported awaits
    else
      cpsBasedAsyncImpl[T](c)(body) // fallback to CPS
  }
}