使用`ActiveRecord with_connection do`和ActionController::Live时出现线程错误

Question

重大修改：自从最初发现这个问题以来，我将其缩小到以下内容。我认为这现在是对该问题的略微更精确的描述。因此，对于原始帖子上的评论可能不完全相关。

编辑：轻微修改后发布在rails/puma项目中：https://github.com/rails/rails/issues/21209，https://github.com/puma/puma/issues/758

编辑：现在使用OS X和Rainbows也可以复制此问题。

摘要：使用Puma并运行长时间连接时，我持续收到与ActiveRecord连接跨线程有关的错误。这表现为像message type 0x## arrived from server while idle和锁定（崩溃）服务器的消息。

设置：

• Ubuntu 15 / OSX Yosemite
• PostgreSQL (9.4) / MySQL (mysqld 5.6.25-0ubuntu0.15.04.1)
• Ruby - MRI 2.2.2p95 (2015-04-13 revision 50295) [x86_64-linux] / Rubinius rbx-2.5.8
• Rails (4.2.3, 4.2.1)
• Puma (2.12.2, 2.11)
• pg (pg-0.18.2) / mysql2

请注意，并非尝试所有上述版本的组合。首先列出的版本是我当前正在测试的版本。

• rails new issue-test
• 添加路由get 'events' => 'streaming#events'
• 添加控制器streaming_controller.rb
• 设置数据库相关内容（pool: 2，但使用不同的池大小也可以看到）

代码：

class StreamingController < ApplicationController

  include ActionController::Live

  def events
    begin
      response.headers["Content-Type"] = "text/event-stream"
      sse = SSE.new(response.stream)
      sse.write( {:data => 'starting'} , {:event => :version_heartbeat})
      ActiveRecord::Base.connection_pool.release_connection
      while true do
        ActiveRecord::Base.connection_pool.with_connection do |conn|
          ActiveRecord::Base.connection.query_cache.clear
          logger.info 'START'
          conn.execute 'SELECT pg_sleep(3)'
          logger.info 'FINISH'
          sse.write( {:data => 'continuing'}, {:event => :version_heartbeat})
          sleep 0.5
         end
      end
    rescue IOError
    rescue ClientDisconnected
    ensure
      logger.info 'Ensuring event stream is closed'
      sse.close
    end
    render nothing: true
  end
end

Puma 配置：

workers 1
threads 2, 2
#...
bind "tcp://0.0.0.0:9292"

#...
activate_control_app

on_worker_boot do
  require "active_record"
  ActiveRecord::Base.connection.disconnect! rescue ActiveRecord::ConnectionNotEstablished
  ActiveRecord::Base.establish_connection(YAML.load_file("#{app_dir}/config/database.yml")[rails_env])
end

• 运行服务器 puma -e production -C path/to/puma/config/production.rb

测试脚本：

#!/bin/bash

timeout 30 curl -vS http://0.0.0.0/events &
timeout 5 curl -vS http://0.0.0.0/events &
timeout 30 curl -vS http://0.0.0.0/events

这通常会导致应用程序服务器完全锁定(在PostgreSQL中，见注释)。可怕的消息来自libpq：

message type 0x44 arrived from server while idle
message type 0x43 arrived from server while idle
message type 0x5a arrived from server while idle
message type 0x54 arrived from server while idle

在“现实世界”中，我有相当多的额外元素，并且问题随机出现。我的研究表明，此消息来自libpq，并为“通信问题，可能使用不同线程中的连接”的子文本。最后，在撰写此文时，我遇到了服务器锁定，但没有任何日志中的单个消息。

所以，问题是：

1. 我遵循的模式在某种程度上是否不合法？我错过或误解了什么？
2. 在这里处理数据库连接的“标准”是什么，应该避免这些问题？
3. 您能看到一种可靠地重现这个问题的方法吗？

或者

1. 这里的潜在问题是什么，我该如何解决？

---

MySQL

如果运行MySQL，则消息略有不同，并且应用程序会恢复（尽管我不确定它是否处于某种未定义状态）：

F, [2015-07-30T14:12:07.078215 #15606] FATAL -- : 
ActiveRecord::StatementInvalid (Mysql2::Error: This connection is in use by: #<Thread:0x007f563b2faa88@/home/dev/.rbenv/versions/2.2.2/lib/ruby/gems/2.2.0/gems/actionpack-4.2.3/lib/action_controller/metal/live.rb:269 sleep>: SELECT  `tasks`.* FROM `tasks`  ORDER BY `tasks`.`id` ASC LIMIT 1):

- button

---

从ActiveRecord的文档中，使用with_connection依赖于传递的块完成。您确定它已经完成了吗？使用Base.connection或checkout处理连接如何？ - Grasshopper

@Grasshopper - 谢谢！我担心这会在请求的生命周期（几个小时）内保持连接打开，因此很快就会耗尽我的连接池？我想，它可能不会完成的方式是如果 sse.write 由于某种原因被阻塞并且线程只是停在那里，即使连接已经断开并且由于某种原因没有返回？（也就是说，我不确定这是否完全解释了来自 libpq 的基于线程问题的消息）。 （将在该方向上尝试一些事情的实验） - button

另一个问题是：你真的需要初始化连接吗？因为从 AR 文档中看来，User.find 似乎已经为你完成了这个任务。 - Grasshopper

不幸的是，由于长时间运行的连接，AR无法独立完成工作，需要一些提示。我注释了with_connection部分，并且能够得到相同的问题。删除release_connection只意味着当我打开与池大小相同的浏览器时，池限制就会达到（这并不理想！）。我不禁感到这可能是Puma、Rails、AR和Ruby之间某个地方的错误/不兼容性问题。类似于线程处理方式或类似问题。 - button

为什么不将所有这些东西移到后台，并将“_user_id”推入队列（可以是redis或rabbitmq），然后后台的某个东西会从中消耗，我不确定您是否想要根据Web请求打开和关闭连接，否则您很快就会用完套接字。 - bjhaid

Answer 1

警告：将“答案”读作“似乎有所不同”

---

如果我将控制器块更改为以下内容，我就不会看到问题发生：

begin
  #...
  while true do
    t = Thread.new do #<<<<<<<<<<<<<<<<<
        ActiveRecord::Base.connection_pool.with_connection do |conn|
            #...
        end
     end
     t.join #<<<<<<<<<<<<<<<<<
  end
  #...
rescue IOError
#...

但我不知道这是否真正解决了问题，还是只是使它极不可能发生。我也不能真正理解为什么这会有所不同。

如果这有所帮助，发布此解决方案，但仍在继续研究该问题。

- button

---

我也有一个长时间运行的进程，这就是我如何使用它的方式。我的问题是，为什么你要为每个迭代启动一个线程？难道不应该在循环外部启动线程并检查每个客户端的连接吗？ - Carlos Roque