Class: Sidekiq::ProcessManager::Manager

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/process_manager/manager.rb

Overview

Process manager for sidekiq. This class is responsible for starting and monitoring that the specified number of sidekiq processes are running. It will also forward signals sent to the main process to the child processes.

Constant Summary collapse

SIGNALS =

Signals that are trapped by the process manager and forwarded to the child processes.

[:INT, :TERM, :USR1, :USR2, :TSTP, :TTIN].freeze
DEFAULT_MEMORY_CHECK_INTERVAL =

Number of seconds between memory checks when a max memory is set.

60

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(process_count: 1, prefork: false, preboot: nil, max_memory: nil, memory_check_interval: nil, mode: nil, silent: false) ⇒ Manager

Create a new process manager.

Parameters:

  • process_count (Integer) (defaults to: 1) —

    The number of sidekiq processes to start.

  • prefork (Boolean) (defaults to: false) —

    If true, the process manager will load the application before forking.

  • preboot (String) (defaults to: nil) —

    If set, the process manager will require the specified file before forking the child processes.

  • memory_check_interval (Numeric) (defaults to: nil) —

    The number of seconds between memory checks when a max memory is set.

  • mode (Symbol) (defaults to: nil) —

    If set to :testing, the process manager will use a mock CLI.

  • silent (Boolean) (defaults to: false) —

    If true, the process manager will not output any messages.

Raises:

  • (ArgumentError) 


30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/sidekiq/process_manager/manager.rb', line 30

def initialize(process_count: 1, prefork: false, preboot: nil, max_memory: nil, memory_check_interval: nil, mode: nil, silent: false)
  require "sidekiq/cli"

  # Get the number of processes to fork
  @process_count = process_count
  raise ArgumentError.new("Process count must be greater than 0") if @process_count < 1

  @prefork = prefork && process_count > 1
  @preboot = preboot if process_count > 1 && !prefork
  @max_memory = ((max_memory.to_i > 0) ? max_memory.to_i : nil)

  if !memory_check_interval.nil? && memory_check_interval.to_f <= 0
    raise ArgumentError.new("Memory check interval must be greater than 0")
  end

  if mode == :testing
    require_relative "../../../spec/support/mocks"
    @cli = MockSidekiqCLI.new(silent)
    @memory_check_interval = (memory_check_interval || 1).to_f
    @exit_grace_period = 0.5
  else
    @cli = Sidekiq::CLI.instance
    @memory_check_interval = (memory_check_interval || DEFAULT_MEMORY_CHECK_INTERVAL).to_f
    @exit_grace_period = 5
  end

  @silent = silent
  @pids = []
  @started = false
  @mutex = Mutex.new
end

Instance Attribute Details

#cli ⇒ Object (readonly)

Returns the value of attribute cli.



19
20
21
 
# File 'lib/sidekiq/process_manager/manager.rb', line 19

def cli
  @cli
end

Instance Method Details

#pids ⇒ Array<Integer>

Get all chile process pids.

Returns:

  • (Array<Integer>) 


160
161
162
 
# File 'lib/sidekiq/process_manager/manager.rb', line 160

def pids
  @mutex.synchronize { @pids.dup }
end

#start ⇒ void

This method returns an undefined value.

Start the process manager. This method will start the specified number of sidekiq processes and monitor them. It will only exit once all child processes have exited. If a child process dies unexpectedly, it will be restarted.

Child processes are manged by sending the signals you would normally send to a sidekiq process to the process manager instead.



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/sidekiq/process_manager/manager.rb', line 71

def start
  raise "Process manager already started" if started?
  @started = true

  load_sidekiq

  master_pid = ::Process.pid

  @signal_pipe_read, @signal_pipe_write = IO.pipe

  @signal_thread = Thread.new do
    Thread.current.name = "signal-handler"

    while @signal_pipe_read.wait_readable
      begin
        signal = @signal_pipe_read.gets
        break if signal.nil?

        signal = signal.strip
        send_signal_to_children(signal.to_sym)
      rescue => e
        log_warning("Error sending signal #{signal} to child processes: #{e.message}")
      end
    end
  end

  # Trap signals that will be forwarded to child processes
  SIGNALS.each do |signal|
    ::Signal.trap(signal) do
      if ::Process.pid == master_pid
        signal = :TERM if signal == :INT
        @signal_pipe_write.puts(signal)
      end
    end
  end

  # Ensure that child processes receive the term signal when the master process exits.
  at_exit do
    if ::Process.pid == master_pid
      if @process_count > 0
        send_signal_to_children(:TERM)
      end
      wait_for_children_to_exit
      log_info("Process manager exiting")
    end
  end

  GC.start
  GC.compact if GC.respond_to?(:compact)
  # I'm not sure why, but running GC operations blocks until we try to write some I/O.
  File.write(File::NULL, "0")

  @process_count.times do
    start_child_process!
  end

  start_memory_monitor if @max_memory

  log_info("Process manager started")
  monitor_child_processes
end

#started? ⇒ Boolean

Return true if the process manager has started.

Returns:

  • (Boolean) 


167
168
169
 
# File 'lib/sidekiq/process_manager/manager.rb', line 167

def started?
  @started
end

#stop ⇒ void

This method returns an undefined value.

Helper to gracefully stop all child processes.



150
151
152
153
154
155
 
# File 'lib/sidekiq/process_manager/manager.rb', line 150

def stop
  stop_memory_monitor
  @process_count = 0
  send_signal_to_children(:TSTP)
  send_signal_to_children(:TERM)
end

#wait(timeout = 5) ⇒ void

This method returns an undefined value.

Helper to wait on the manager to wait on child processes to start up.

Parameters:

  • timeout (Integer) (defaults to: 5) —

    The number of seconds to wait for child processes to start.

Raises:

  • (Timeout::Error) 


137
138
139
140
141
142
143
144
145
 
# File 'lib/sidekiq/process_manager/manager.rb', line 137

def wait(timeout = 5)
  timeout_time = monotonic_time + timeout
  while monotonic_time <= timeout_time
    return if pids.size == @process_count
    sleep(0.01)
  end

  raise Timeout::Error.new("child processes failed to start in #{timeout} seconds")
end