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

Question

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

场景

1. 任务并行执行
2. 如果失败超过X次
3. 停止所有尚未完成状态的任务(越快越好)

我的解决方案

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

实验

关于 Ammonite回复

{
  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)

当您只有一个(或几个)线程时，“守卫”任务需要一段时间才能进行另一轮检查。使用 Task.gather，您可以同时启动 100 个任务，因此调度程序非常繁忙，“守卫”无法再次检查，直到其他任务已经完成。如果每个任务只有一个线程，则调度程序无法保证公平性，因此“守卫”不公平地检查得更频繁并且可以更快完成。

如果您使用 Task.sequence，那 100 个任务将按顺序执行，这就是为什么“守卫”任务有更多机会在需要时尽快完成的原因。如果你想让你的代码保持原样，你可以使用 Task.gatherN(parallelism = 4) 这将限制并行度，因此允许你的“守卫”更频繁地检查(介于两者之间的中间地带) Task.sequence 和 Task.gather)。

在我看来它有点像 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，我们可以将其与 任务。出于演示目的，我还将阈值更改为 30，只有其他所有任务都会“出错”。因此无论线程池大小如何，预期的输出始终在 completed tasks: 60, errors: 30 左右。

我仍在使用 errCounter.get.restartUntil(_ >= threshold) 进行轮询，根据我的口味，这可能会消耗过多的 CPU，但它接近您最初的想法并且运行良好。

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

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

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