如何从Async[IO]创建Async[Future]

Question

我试图在我的代码中为doobie仓库隐式地添加异步和同步。同步和AsyncF IO工作正常。我想把它们转化为未来和面临的问题

我尝试从IO创建我自己的Aync

def futureAsync(implicit F: MonadError[Future, Throwable]): Async[Future] = new Async[Future] {
    override def async[A](k: (Either[Throwable, A] => Unit) => Unit): Future[A] = IO.async(k).unsafeToFuture()

    override def asyncF[A](k: (Either[Throwable, A] => Unit) => Future[Unit]): Future[A] =
      throw new Exception("Not implemented Future.asyncF")

    override def suspend[A](thunk: => Future[A]): Future[A] = thunk

    override def bracketCase[A, B](acquire: Future[A])(use: A => Future[B])(release: (A, ExitCase[Throwable]) => Future[Unit]): Future[B] =
      throw new Exception("Not implemented Future.bracketCase")

    override def raiseError[A](e: Throwable): Future[A] = F.raiseError(e)

    override def handleErrorWith[A](fa: Future[A])(f: Throwable => Future[A]): Future[A] = F.handleErrorWith(fa)(_ => f(new Exception("")))

    override def pure[A](x: A): Future[A] = F.pure(x)

    override def flatMap[A, B](fa: Future[A])(f: A => Future[B]): Future[B] = F.flatMap(fa)(f)

    override def tailRecM[A, B](a: A)(f: A => Future[Either[A, B]]): Future[B] = F.tailRecM(a)(f)
  }

我被其中的两个函数asyncF和bracketCase的实现所震撼，有人能帮上忙吗？

Answer 1

正如Reactormonk在上面的评论中所说的那样，不可能为Future编写一个具有正确语义的Async实例，因为Async扩展了Sync，而Sync需要一个可以重复运行的计算表示，而Scala的未来在它们被定义时就开始运行，不能重新运行。

一个非法的实例

不过，亲眼看看这一点很有指导意义，我鼓励您尝试编写自己的可编译但(必然)非法的Async[Future]实例，而不必查看下一段代码。不过，为了这个例子，这里有一个我头顶上的快速草图：

import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Failure, Success}
import cats.effect.{Async, ExitCase, IO}

def futureAsync(implicit c: ExecutionContext): Async[Future] = new Async[Future] {
  def async[A](k: (Either[Throwable, A] => Unit) => Unit): Future[A] =
    IO.async(k).unsafeToFuture()

  def asyncF[A](k: (Either[Throwable, A] => Unit) => Future[Unit]): Future[A] = {
    val p = Promise[A]()
    val f = k {
      case Right(a) => p.success(a)
      case Left(e) => p.failure(e)
    }
    f.flatMap(_ => p.future)
  }

  def suspend[A](thunk: => Future[A]): Future[A] = Future(thunk).flatten

  def bracketCase[A, B](acquire: Future[A])(use: A => Future[B])(
    release: (A, ExitCase[Throwable]) => Future[Unit]
  ): Future[B] = acquire.flatMap { a =>
    use(a).transformWith {
      case Success(b) => release(a, ExitCase.Completed).map(_ => b)
      case Failure(e) => release(a, ExitCase.Error(e)).flatMap(_ => Future.failed(e))
    }
  }

  def raiseError[A](e: Throwable): Future[A] = Future.failed(e)
  def handleErrorWith[A](fa: Future[A])(f: Throwable => Future[A]): Future[A] =
    fa.recoverWith { case t => f(t) }

  def pure[A](x: A): Future[A] = Future.successful(x)
  def flatMap[A, B](fa: Future[A])(f: A => Future[B]): Future[B] = fa.flatMap(f)
  def tailRecM[A, B](a: A)(f: A => Future[Either[A, B]]): Future[B] = f(a).flatMap {
    case Right(b) => Future.successful(b)
    case Left(a) => tailRecM(a)(f)
  }
}

这将编译得很好，并且可能在某些情况下有效(但请不要实际使用它！)我们已经说过它不可能有正确的语义，但是我们可以通过使用can effect的法则模块来说明这一点。

查证法律

首先，我们需要一些你并不真正需要担心的东西：

import cats.kernel.Eq, cats.implicits._
import org.scalacheck.Arbitrary

implicit val throwableEq: Eq[Throwable] =  Eq.by[Throwable, String](_.toString)
implicit val nonFatalArbitrary: Arbitrary[Throwable] =
  Arbitrary(Arbitrary.arbitrary[Exception].map(identity))

implicit def futureEq[A](implicit A: Eq[A], ec: ExecutionContext): Eq[Future[A]] =
  new Eq[Future[A]] {
    private def liftToEither(f: Future[A]): Future[Either[Throwable, A]] =
      f.map(Right(_)).recover { case e => Left(e) }

      def eqv(fx: Future[A], fy: Future[A]): Boolean =
        scala.concurrent.Await.result(
        liftToEither(fx).zip(liftToEither(fy)).map {
          case (rx, ry) => rx === ry
        },
        scala.concurrent.duration.Duration(1, "second")
      )
  }

然后，我们可以为我们的实例定义一个检查Async定律的测试：

import cats.effect.laws.discipline.{AsyncTests, Parameters}
import org.scalatest.FunSuite
import org.typelevel.discipline.scalatest.Discipline

object FutureAsyncSuite extends FunSuite with Discipline {
  implicit val ec: ExecutionContext = ExecutionContext.global

  implicit val params: Parameters =
    Parameters.default.copy(allowNonTerminationLaws = false)

  checkAll(
    "Async",
    AsyncTests[Future](futureAsync).async[String, String, String]
  )
}

然后我们可以运行法律测试：

scala> FutureAsyncSuite.execute()
FutureAsyncSuite:
- Async.async.acquire and release of bracket are uncancelable
- Async.async.ap consistent with product + map
- Async.async.applicative homomorphism
...

您将看到大多数测试都是绿色的；这个实例可以做很多正确的事情。

触犯法律的地方

但是，它确实显示了三个失败的测试，包括以下内容：

- Async.async.repeated sync evaluation not memoized *** FAILED ***
  GeneratorDrivenPropertyCheckFailedException was thrown during property evaluation.
   (Discipline.scala:14)
    Falsified after 1 successful property evaluations.
    Location: (Discipline.scala:14)
    Occurred when passed generated values (
      arg0 = "淳칇멀",
      arg1 = org.scalacheck.GenArities$$Lambda$7154/1834868832@1624ea25
    )
    Label of failing property:
      Expected: Future(Success(驅ṇ숆㽝珅뢈矉))
  Received: Future(Success(淳칇멀))

如果查看laws definitions，您将看到这是一个使用delay定义Future值，然后多次对其进行排序的测试，如下所示：

val change = F.delay { /* observable side effect here */ }
val read = F.delay(cur)

change *> change *> read

另外两个失败是类似的“未记忆”违规。这些测试应该会看到副作用发生两次，但在我们的例子中，不可能以这种方式为Future编写delay或suspend (不过，值得尝试说服自己这种情况)。

你应该怎么做？

总而言之:您可以编写一个Async[Future]实例，它可以通过78个Async法律测试中的75个，但不可能编写一个通过所有测试的实例，使用非法实例是一个非常糟糕的想法:您的代码和库的潜在用户都会假设您的实例是合法的，如果您没有达到这个假设，您就会为复杂和恼人的bug打开大门。

值得注意的是，为Future编写一个具有合法Async实例的最小包装器并不太难(例如，我的catbird库中有一个用于Twitter未来的包装器，名为Rerunnable )。不过，您真的应该坚持使用cats.effect.IO，并在使用传统Future-based API的代码的任何部分中使用提供的转换来与未来进行转换。