监控 goroutine 和通道泄漏/饥饿？

Question

我有许多工作人员正在运行，将工作项目从队列中拉出。比如：

(def num-workers 100)

(def num-tied-up (atom 0))
(def num-started (atom 0))

(def input-queue (chan (dropping-buffer 200))

(dotimes [x num-workers]
  (go
    (swap num-started inc)
        (loop []
          (let [item (<! input-queue)]
            (swap! num-tied-up inc)
            (try 
              (process-f worker-id item)
              (catch Exception e _))
              (swap! num-tied-up dec))
          (recur)))
          (swap! num-started dec))))

希望num-tied-up 代表在给定时间点执行工作的工人数量。 num-tied-up 的值徘徊在相当一致的50 附近，有时是60。由于num-workers 为100，num-started 的值正如预期的那样为100（即所有go 例程都在运行），这感觉就像有一个舒适的余量。

我的问题是input-queue 正在增长。我预计它会徘徊在零标记附近，因为有足够的工人从它上面取下物品。但在实践中，最终它会最大化并丢弃事件。

看起来tied-up 在num-workers 中有足够的空间，所以工作人员应该可以从队列中取出工作。

我的问题：

1. 我能做些什么来使它更健壮吗？
2. 是否有任何其他诊断程序可以用来找出问题所在？有没有办法监控当前工作的 goroutines 的数量，以防它们死亡？
3. 您能否使观察结果与数据相符？

Answer 1

go 例程完成的工作不应执行任何 IO 或阻塞操作（如线程/睡眠），因为所有 go 例程共享同一个线程池，该线程池现在具有固定大小 cpus * 2 + 42。对于 IO 有界工作，请使用 core.async/thread。

请注意，线程池限制了将同时执行的 go 例程的数量，但您可以有很多等待执行的时间。

打个比方，如果你启动 Chrome、Vim 和 iTunes（相当于 3 个 go 例程）但你的笔记本电脑中只有一个 CPU（相当于一个大小为 1 的线程池），那么它们中只有一个会执行在 CPU 中，另一个将等待执行。操作系统负责暂停/恢复程序，因此看起来它们都在同时运行。 Core.async 只是做同样的事情，但不同的是，core.async 可以在 go-routines 遇到 ! 时暂停它们。

现在回答你的问题：

1. 没有。尝试/捕获所有异常是您的最佳选择。另外，监控队列的大小。我会重新评估 core.async 的需求。 Zach Tellman 有一个很好的 thread pool implementation，有很多指标。
2. 线程转储将显示所有 core.async 线程阻塞的位置，但正如我所说，您不应该在 go-routine 线程中执行任何 IO 工作。看看core.async/pipeline-blocking
3. 如果您有 4 个内核，您将获得 50 个 core.async 线程池，这与您对 50 个并发 go 块的观察结果相匹配。其他 50 个 go 块正在运行，但要么等待工作出现在队列中，要么等待执行时间段。请注意，它们都有机会至少执行一次，因为 num-started 为 100。

希望对你有帮助。

Answer 2

当我修复代码的缩进时，我看到的是这样的：

(def num-workers 100)

(def num-tied-up (atom 0))
(def num-started (atom 0))

(def input-queue (chan (dropping-buffer 200))

(dotimes [x num-workers]
  (go
    (swap num-started inc)
    (loop []
      (let [item (<! input-queue)]
        (swap! num-tied-up inc)
        (try 
          (process-f worker-id item)
          (catch Exception e _))
          (swap! num-tied-up dec))
      (recur)))
  (swap! num-started dec))

)) ;; These parens don't balance.

忽略额外的括号，我认为这是一些复制/粘贴错误，这里有一些观察：

1. 您在 go 线程内增加 num-started，但在创建该线程后立即在 go 线程外减少它。递减很可能总是在递增之前发生。
2. 您创建的 100 个循环（每个 go 线程一个）永远不会终止。这本身不是问题，只要这是有意设计的。

请记住，生成 go 线程并不意味着当前线程（执行生成的线程，dotimes 在其中执行）将阻塞。我可能弄错了，但看起来您的代码假设(swap! num-started dec) 只会在紧接其上方生成的 go 线程完成时运行。但这不是真的，即使您的 go 线程 确实 最终完成（如上所述，它们没有完成）。