Usage with other effect systems besides Future

[jimmydivvy]

The README indicates that this library can be used with other implementations of the future pattern - however I couldn't find documentation on how to do this.

In particular I'm interested in using with scalaz Task. Is this possible?

[jvican]

I am also very interested in this. Up +1

[retronym]

See:

https://github.com/scala/async/blob/master/src/main/scala/scala/async/internal/ScalaConcurrentAsync.scala

https://github.com/scala/async/blob/master/src/main/scala/scala/async/internal/FutureSystem.scala#L79-L138

https://github.com/scala/async/blob/master/src/main/scala/scala/async/Async.scala#L45

For the spots you would need to replicate to add an adapter for Task. You might need to model the absense of an execution context on Task as a Unit typed. You can see this in the example of the "not-actually-concurrent future system" in https://github.com/scala/async/blob/master/src/main/scala/scala/async/internal/AsyncId.scala

[SethTisue]

would someone like to turn this into a pull request with a small addition to the doc?

[adamw]

I think the major problem here would be that the FutureSystem assumes the presence of a Promise concept, and as far as I know the Task in scalaz, or IO in cats don't have that.

However, there's https://github.com/pelotom/effectful, which seems to do the same thing for scalaz, or is it somehow different?

[dsilvasc]

@adamw effectful is effectively abandoned: pelotom/effectful#15

[adamw]

@dsilvasc True, though there's also https://github.com/monadless/monadless (not sure if it's maintained, but it usually works good enough)

[dsilvasc]

@adamw There's a question about monadless support for scala 2.13 from over 2 months ago with no response:
monadless/monadless#9

Don't know about maintenance plans for Stateless Future:
https://github.com/qifun/stateless-future

ThoughtWorks Each doesn't seem to have a scala 2.13 version published to Maven Central, but they might actively use it in production with their customers.
https://github.com/ThoughtWorksInc/each

ThoughtWorks seem to be moving on to a compiler plugin though -- same primary author from Each:
https://javadoc.io/page/com.thoughtworks.dsl/dsl_2.12/latest/com/thoughtworks/dsl/index.html
https://github.com/ThoughtWorksInc/Dsl.scala
https://github.com/ThoughtWorksInc/Dsl.scala/wiki/Benchmarks:-Dsl.scala-vs-Monix-vs-Cats-Effect-vs-Scalaz-Concurrent-vs-Scala-Async-vs-Scala-Continuation

Sorry for the tangent -- I think that's all of the related projects :)

[adamw]

I would be more concerned about unmerged PRs than issues without comment ;) Anyway, all of these seem rather abandoned. Would be nice of course to combine all these efforts into a single one, working in any Cats/Scalaz monad (like Kotlin's coroutines), but ... there's a finite amount of time ;)

[dsilva]

@SethTisue @retronym @adamw

I took a look at ScalaConcurrentAsync and AsyncId, then took a stab at implementing an adapter for a Task-like abstraction where Task[T] is a wrapper around a memoized function (ExecutionContext => Future[T]) and with an API similar to that of Future (but without an implicit ExecutionContext parameter everywhere). Here's what a naive translation would look like:

import scala.async.internal.{AsyncBase, FutureSystem}
import scala.concurrent.Promise
import scala.reflect.macros.whitebox

import scala.language.experimental.macros

object TaskAsync extends AsyncBase {
  lazy val futureSystem = TaskFutureSystem
  type FS = TaskFutureSystem.type

  def async[T](body: => T): Task[T] = macro asyncTaskImpl[T]

  def asyncTaskImpl[T: c.WeakTypeTag](c: whitebox.Context)(body: c.Expr[T]): c.Expr[Task[T]] = {
    val u: c.Expr[Unit] = c.Expr[Unit](c.parse("()"))
    asyncImpl[T](c)(body)(u)
  }
}

object TaskFutureSystem extends FutureSystem {
  override type Prom[A] = Promise[A]
  override type Fut[A] = Task[A]
  override type ExecContext = Unit
  override type Tryy[A] = scala.util.Try[A]

  override def mkOps(c0: whitebox.Context): Ops { val c: c0.type } = new Ops {
    val c: c0.type = c0
    import c.universe._

    def promType[A](implicit evidence: c.universe.WeakTypeTag[A]): c.universe.Type = weakTypeOf[Prom[A]]
    def tryType[A](implicit evidence: c.universe.WeakTypeTag[A]): c.universe.Type = weakTypeOf[Tryy[A]]

    def execContextType: Type = weakTypeOf[ExecContext]

    def createProm[A](implicit evidence: c.universe.WeakTypeTag[A]): c.universe.Expr[Promise[A]] = reify {
      Promise[A]()
    }

    def promiseToFuture[A: WeakTypeTag](prom: Expr[Prom[A]]): c.universe.Expr[Task[A]] = reify {
      new Task(_ => prom.splice.future)
    }

    def future[A: WeakTypeTag](a: Expr[A])(execContext: Expr[ExecContext]): c.universe.Expr[Task[A]] = reify {
      new Task(implicit ec => Future { a.splice })
    }

    def onComplete[A, B](
        task: Expr[Fut[A]],
        fun: Expr[scala.util.Try[A] => B],
        execContext: Expr[ExecContext]): Expr[Unit] = reify {
      task.splice.onComplete(fun.splice)
    }

    override def continueCompletedFutureOnSameThread: Boolean = true

    override def getCompleted[A: WeakTypeTag](task: Expr[Fut[A]]): Expr[Tryy[A]] = reify {
      if (task.splice.isCompleted) future.splice.value.get else null
    }

    def completeProm[A](prom: Expr[Prom[A]], value: Expr[scala.util.Try[A]]): Expr[Unit] = reify {
      prom.splice.complete(value.splice)
      c.Expr[Unit](Literal(Constant(()))).splice
    }

    def tryyIsFailure[A](tryy: Expr[scala.util.Try[A]]): Expr[Boolean] = reify {
      tryy.splice.isFailure
    }

    def tryyGet[A](tryy: Expr[Tryy[A]]): Expr[A] = reify {
      tryy.splice.get
    }
    def tryySuccess[A: WeakTypeTag](a: Expr[A]): Expr[Tryy[A]] = reify {
      scala.util.Success[A](a.splice)
    }
    def tryyFailure[A: WeakTypeTag](a: Expr[Throwable]): Expr[Tryy[A]] = reify {
      scala.util.Failure[A](a.splice)
    }
  }
}

This doesn't work because the async macro expects the implementation of the future method to generate an expression that kicks off execution, and tasks don't start running when they're created.

To clarify, here's an example:

    val runCount = new AtomicInteger(0)

    val t = async {
      val task = async {
        runCount.incrementAndGet()
        5
      }

      assert(runCount.get == 0)

      val ten = await(task) * 2
      assert(ten == 10)

      assert(runCount.get == 1)

      ten
    }

    val future = t.run()(ExecutionContext.global)
    val result = Await.result(future, Duration.Inf)

    assert(runCount.get == 1)
    assert(result == 10)

and the macro expansion printed by -Ymacro-debug-lite and some manual cleanup for legibility:

    val runCount = new AtomicInteger(0)

    class StateMachine extends AnyRef with (snapchat.concurrent.TaskFutureSystem.Tryy[Any] => Unit) with (() => Unit) {
      private[this] var await$macro$4$macro$6: Int = _
      private[this] var state: Int = 0
      /*private[this] */
      val result: snapchat.concurrent.TaskFutureSystem.Prom[Int] = Promise.apply[Int]()
      //   def result: snapchat.concurrent.TaskFutureSystem.Prom[Int] = result
      private[this] val execContext: Unit = ()
      //   def execContext: Unit = execContext

      def apply(tr: snapchat.concurrent.TaskFutureSystem.Tryy[Int]): Unit = while$macro$8() {
        try {
          state match {
            case 0 =>
              val task = new Task(implicit ec => Future {
                runCount.incrementAndGet()
                5
              })
              assert(runCount.get().==(0))
              val awaitable$macro$3
                : snapchat.concurrent.Task[Int] /* @scala.reflect.internal.annotations.uncheckedBounds */ =
                task
              state = 1
              val completed$macro$7: snapchat.concurrent.TaskFutureSystem.Tryy[Int] =
                if (awaitable$macro$3.isCompleted)
                  awaitable$macro$3.value.get
                else
                  null
              if (null.ne(completed$macro$7)) {
                if (completed$macro$7.isFailure) {
                  /*stateMachine$macro$2.this.*/
                  result.complete(completed$macro$7)
                  return ()
                } else {
                  await$macro$4$macro$6 = completed$macro$7.get
                  state = 2
                }
                ()
              } else {
                awaitable$macro$3.onComplete(this)
                return ()
              }
              ()

            case 2 => {
              /*stateMachine$macro$2.this.*/
              result.complete(Success({
                val x$macro$5: Int = 2
                val ten: Int = await$macro$4$macro$6.*(x$macro$5)
                assert(ten.==(10))
                assert(runCount.get().==(0))
                ten
              }))
              ()
            }
            return ()

          case 1 =>
              if (tr.isFailure) {
                /*stateMachine$macro$2.this.*/
                result
                  .complete(tr)
                return ()
              } else {
                await$macro$4$macro$6 = tr.get
                state = 2
              }
              ()

            case _ => throw new IllegalStateException()
          }
        } catch {
          case throwable @ (_: Throwable) =>
            result.complete(Failure[Int](throwable))
            return ()
        }
        while$macro$8()
      }
      def apply(): Unit = StateMachine.this.apply(null)
    }

    val stateMachine = new StateMachine

    new Task(implicit ec => Future { stateMachine.apply() })
    new Task(_ => stateMachine$macro$2.result.future)

Note the last two lines. The call to stateMachine.apply() is wrapped in a task that's never started, then discarded. The other task (on the last line) grabs the future from the promise that was created by new StateMachine but that won't ever be completed because stateMachine.apply() never actually runs.

[adamw]

@dsilva I ended up using https://github.com/monadless/monadless, which has the same functionality, but is general and works for any monad. Unfortunately, it's also unmaintained.