Rails delayed_jobのworker数を変更してrestartするときの注意

delayed_jobのworker数を変更してrestartすると前に起動していたプロセスが消えない

[現象]Capistranoでdeployする時にdelayed_jobのワーカー数を変更すると古いワーカーが残ってしまう

最初にワーカーを1個にしてdeployしておいて、処理数が足りなくなってワーカーを2個に増やした場合、

delayed_job.0 (新)
delayed_job.1 (新)
delayed_job (古)

という状態になる。

ちなみにワーカーを3個から2個に減らすと

delayed_job.0 (新)
delayed_job.1 (新)
delayed_job.2 (古)

となる。

[原因]number_of_workersオプションの挙動

number_of_workersオプション無しで実行する

delayed_jobというプロセスが起動する

$ bundle exec bin/delayed_job start
delayed_job: process with pid 7665 started.

$ ps aux | grep [d]elayed_job
ubuntu    7665 32.9  7.9 731312 322824 ?       Sl   15:42  11:41 delayed_job

number_of_workersオプションありで実行する

suffixが付く

$ bundle exec bin/delayed_job --number_of_workers 2 start
Use "rmagick" instead
delayed_job.0: process with pid 7881 started.
delayed_job.1: process with pid 7887 started.

$ ps aux | grep [d]elayed_job
ubuntu    7881 33.9  5.4 564336 220212 ?       Sl   16:19   0:04 delayed_job.0                                   
ubuntu    7887 31.4  5.4 564692 220324 ?       Sl   16:19   0:04 delayed_job.1

オプションありで実行したあとに、オプション無しでrestartしてみる

suffixが付いているプロセスは生きたままで、新たにdelayed_jobというワーカーが生まれる。

$ bundle exec bin/delayed_job restart
delayed_job: warning: no instances running. Starting...
delayed_job: process with pid 7939 started.

$ ps aux | grep [d]elayed_job
ubuntu    7881 42.2  7.6 653708 309632 ?       Sl   16:19   0:46 delayed_job.0                                   
ubuntu    7887 41.9  7.6 653896 309740 ?       Sl   16:19   0:45 delayed_job.1                                   
ubuntu    7939 41.4  5.5 568964 224836 ?       Sl   16:21   0:12 delayed_job

ソース

ここ。
1個以上の場合にはプロセス数にsuffixを付ける。
stopする時には、stopの時に渡されたnumber_of_workersからプロセスを探すので、stopから漏れるワーカーが生まれる。
https://github.com/collectiveidea/delayed_job/blob/master/lib/delayed/command.rb#L97

Capistranoでの対策

capistranoでdeployする時の対応

config/deploy.rb内で、今起動しているワーカーの全pidを探して来てTERMを送るようにする。
これでうまく動いた。
ソースの一部は、daemonsというgemでワーカーをstopする時のものを拝借。

config/deploy.rb

(snip)
  task :stop_all_delayed_job do
    on roles(:worker) do
      within release_path do
        begin
          pids = capture(:cat, "tmp/pids/delayed_job*")
        rescue
          info "there is no delayed_job process."
          next
        end
        pids.split("\n").each do |pid_s|
          pid = pid_s.to_i
          begin
            execute(:kill, '-s', 'TERM', pid)
            Timeout.timeout(5, TimeoutError) do
              while (execute(:kill, '-s', 'EXIT', pid) rescue false)
                 sleep(0.1)
              end
            end
          rescue ::Exception => e
            puts "exception while trying to stop monitor process #{pid}: #{e}"
          end
        end
        begin; File.delete("tmp/pids/delayed_job*"); rescue ::Exception; end
      end
    end
  end
  
  after 'restart', 'stop_all_delayed_job'
  after 'stop_all_delayed_job', 'delayed_job:start'
(/snip)

まとめ

みんな困ってるんじゃないかと思ったので、公式に解決策があるんじゃないかと思いつつもこれで解決。