故障超过阈值时停止所有异步任务？

Question

我正在使用 Monix Task 进行异步控制。

场景

任务并行执行
如果故障发生超过 X 次
停止所有尚未完成状态的任务（越快越好）

我的解决方案

我想到了在 1.result 和 2.error counter 之间竞赛，并取消失败者的想法。
通过 Task.race 如果错误计数器首先达到阈值，则任务将被 Task.race 取消。

实验

{
  import $ivy.`io.monix::monix:3.1.0`
  import monix.eval.Task
  import monix.execution.atomic.Atomic
  import scala.concurrent.duration._
  import monix.execution.Scheduler
  //import monix.execution.Scheduler.Implicits.global
  implicit val s = Scheduler.fixedPool("race", 2) // pool size

  val taskSize = 100
  val errCounter = Atomic(0)
  val threshold = 3

  val tasks = (1 to taskSize).map(_ => Task.sleep(100.millis).map(_ => errCounter.increment()))
  val guard = Task(f"stop because too many error: ${errCounter.get()}")
    .restartUntil(_ => errCounter.get() >= threshold)

  val race = Task
    .race(guard, Task.gather(tasks))
    .runToFuture
    .onComplete { case x => println(x); println(f"completed task: ${errCounter.get()}") }
}

问题

结果取决于线程池大小！？

池大小 1
结果几乎总是任务成功，即没有停止。

Success(Right(.........))
completed task: 100 // all task success !

池大小 2
成功与失败之间非常不确定，取消也不准确。例如：

Success(Left(stop because too many error: 1))
completed task: 98

最迟在完成 98 个任务后才取消。
错误计数小到阈值很奇怪。

默认的全局调度程序得到同样的结果行为。

对于池大小 200
它更具确定性，停止时间更早，因此在完成较少任务的意义上更准确。

Success(Left(stop because too many error: 2))
completed task: 8

池大小越大越好。

如果我将 Task.gather 更改为 Task.sequence 执行，所有问题都消失了！

这种对池大小的依赖性的原因是什么？一旦发生太多错误，如何改进它或者是否有更好的方法来停止任务？

Answer 1

您所看到的可能是 monix 调度程序的效果以及它如何实现公平性。这是一个相当复杂的主题，但文档和 scaladocs 非常好（参见：https://monix.io/docs/3x/execution/scheduler.html#execution-model）

当您只有一个（或几个）线程时，需要一段时间才能让 "guard" 任务再次轮到检查。使用 Task.gather 时，您一次启动 100 个任务，因此调度程序非常繁忙，并且 "guard" 在其他任务完成之前无法再次检查。如果每个任务只有一个线程，则调度程序无法保证公平性，因此 "guard" 不公平地检查得更频繁并且可以更快完成。

如果您使用 Task.sequence，那 100 个任务将按顺序执行，这就是 "guard" 任务有更多机会在需要时尽快完成的原因。如果你想让你的代码保持原样，你可以使用 Task.gatherN(parallelism = 4) 这将限制并行性，因此允许你的 "guard" 更频繁地检查（Task.sequence 和 [= 之间的中间地带11=]).

这对我来说有点像 Go 代码（使用 Task.race 就像 Go 的 select），而且您还使用了不受约束的副作用，这进一步使理解正在发生的事情变得复杂。我试图以一种更惯用的方式重写你的程序，对于复杂的并发，我通常会使用像 Observable:

这样的流

import cats.effect.concurrent.Ref
import monix.eval.Task
import monix.execution.Scheduler
import monix.reactive.Observable

import scala.concurrent.duration._

object ErrorThresholdDemo extends App {

  //import monix.execution.Scheduler.Implicits.global
  implicit val s: Scheduler = Scheduler.fixedPool("race", 2) // pool size

  val taskSize  = 100
  val threshold = 30

  val program = for {
    errCounter <- Ref[Task].of(0)

    tasks = (1 to taskSize).map(n => Task.sleep(100.millis).flatMap(_ => errCounter.update(_ + (n % 2))))

    tasksFinishedCount <- Observable
      .fromIterable(tasks)
      .mapParallelUnordered(parallelism = 4) { task =>
        task
      }
      .takeUntilEval(errCounter.get.restartUntil(_ >= threshold))
      .map(_ => 1)
      .sumL

    errorCount <- errCounter.get
    _          <- Task(println(f"completed tasks: $tasksFinishedCount, errors: $errorCount"))
  } yield ()

  program.runSyncUnsafe()
}

如您所见，我不再使用全局可变副作用，而是使用 Ref，它内部也使用 Atomic，但提供了一个功能性的 api，我们可以将其与 Task。出于演示目的，我还将阈值更改为 30，并且只有每个其他任务都会 "error"。因此，无论线程池大小如何，预期输出始终在 completed tasks: 60, errors: 30 左右。

我仍在使用 errCounter.get.restartUntil(_ >= threshold) 的轮询，根据我的口味，这可能有点过火 CPU，但它接近您最初的想法并且效果很好。

通常我不会预先创建任务列表，而是将输入放入 Observable 并在 .mapParallelUnordered 中创建任务。此代码保留您的列表，这就是为什么不涉及真正的映射（它已经包含任务）。

您可以像 Task.gatherN 一样选择所需的并行度，我觉得这非常好。

如果还有什么不清楚的地方，请告诉我:)

故障超过阈值时停止所有异步任务？

stop all async Task when they fails over threshold?

multithreading

scala

concurrency

scala-cats

monix

场景

我的解决方案

实验

问题