Class: BackgroundQueue::ServerLib::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/background_queue/server_lib/server.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Server

Returns a new instance of Server.



15
16
17
18
19
20
21
22
23
 
# File 'lib/background_queue/server_lib/server.rb', line 15

def initialize
  @running = false
  @stat_mutex = Mutex.new
  @stats = {
    :tasks=>0,
    :run_tasks=>0,
    :running=>0
  }
end

Instance Attribute Details

#config ⇒ Object

Returns the value of attribute config.



7
8
9
 
# File 'lib/background_queue/server_lib/server.rb', line 7

def config
  @config
end

#event_server ⇒ Object

Returns the value of attribute event_server.



10
11
12
 
# File 'lib/background_queue/server_lib/server.rb', line 10

def event_server
  @event_server
end

#jobs ⇒ Object

Returns the value of attribute jobs.



12
13
14
 
# File 'lib/background_queue/server_lib/server.rb', line 12

def jobs
  @jobs
end

#logger ⇒ Object

Returns the value of attribute logger.



13
14
15
 
# File 'lib/background_queue/server_lib/server.rb', line 13

def logger
  @logger
end

#task_queue ⇒ Object

Returns the value of attribute task_queue.



9
10
11
 
# File 'lib/background_queue/server_lib/server.rb', line 9

def task_queue
  @task_queue
end

#thread_manager ⇒ Object

Returns the value of attribute thread_manager.



8
9
10
 
# File 'lib/background_queue/server_lib/server.rb', line 8

def thread_manager
  @thread_manager
end

#workers ⇒ Object

Returns the value of attribute workers.



11
12
13
 
# File 'lib/background_queue/server_lib/server.rb', line 11

def workers
  @workers
end

Instance Method Details

#change_stat(stat, delta) ⇒ Object 



271
272
273
274
275
 
# File 'lib/background_queue/server_lib/server.rb', line 271

def change_stat(stat, delta)
  @stat_mutex.synchronize {
    @stats[stat] += delta
  }
end

#check_not_running(options) ⇒ Object

Raises:



147
148
149
150
151
 
# File 'lib/background_queue/server_lib/server.rb', line 147

def check_not_running(options)
  proc_id = get_pid(options)
  raise BackgroundQueue::ServerLib::InitError, "Process #{proc_id} already running" unless proc_id.nil?
  nil
end

#daemonize(options) ⇒ Object 



205
206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/background_queue/server_lib/server.rb', line 205

def daemonize(options)
  fork{
    stdin = open '/dev/null', 'r'
    stdout = open '/dev/null', 'w'
    stderr = open '/dev/null', 'w'
    STDIN.reopen stdin
    STDOUT.reopen stdout
    STDERR.reopen stderr
    fork{
      write_pid(options) unless options[:skip_pid]
      run(options)
    } and exit!
  }
end

#get_pid(options) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/background_queue/server_lib/server.rb', line 128

def get_pid(options)
  sPid = nil
  begin
    sPid = File.open(get_pid_path(options)) { |f|
      f.read
    }
  rescue
    return nil
  end
  return nil if sPid.nil? || sPid.to_i == 0
  nPid = sPid.to_i
  begin
    Process.kill(0, nPid)
    return nPid
  rescue
    return nil
  end
end

#get_pid_path(options) ⇒ Object 



120
121
122
123
124
125
126
 
# File 'lib/background_queue/server_lib/server.rb', line 120

def get_pid_path(options)
  if options[:pid_file]
    options[:pid_file]
  else
    "/var/run/background_queue.pid"
  end
end

#get_stats ⇒ Object 



277
278
279
280
281
 
# File 'lib/background_queue/server_lib/server.rb', line 277

def get_stats
  @stat_mutex.synchronize {
     @stats.clone
  }
end

#init_logging(path, level) ⇒ Object 



103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/background_queue/server_lib/server.rb', line 103

def init_logging(path, level)
  unless path.nil? || path.to_s.strip.length == 0
    path = resolve_logging_path(path)
    begin
      @logger = Logger.new(path, "daily")
      set_logging_level(@logger, level)
    rescue Exception=>e
      raise BackgroundQueue::ServerLib::InitError, "Error initializing log file #{path}: #{e.message}"
    end
  end
  if @logger.nil?
    #just make a fallback logger...
    @logger = Logger.new($stderr)
    set_logging_level(@logger, "fatal")
  end
end

#kill_pid(options) ⇒ Object 



170
171
172
173
174
175
176
177
 
# File 'lib/background_queue/server_lib/server.rb', line 170

def kill_pid(options)
  proc_id = get_pid(options)
  begin
    Process.kill(9, proc_id) unless proc_id.nil?
  rescue 
    #dont care... the process may have died already?
  end
end

#load_configuration(path) ⇒ Object 



72
73
74
75
 
# File 'lib/background_queue/server_lib/server.rb', line 72

def load_configuration(path)
  @config = BackgroundQueue::ServerLib::Config.load_file(path)
  true
end

#load_tasks(path) ⇒ Object 



283
284
285
286
287
288
289
290
291
292
293
294
295
 
# File 'lib/background_queue/server_lib/server.rb', line 283

def load_tasks(path)
  return if path.nil?
  if File.exist?(path)
    begin
      File.open(path, 'r') { |io| 
        task_queue.load_from_file(io)
      }
    rescue Exception=>e
      logger.error("Error loading tasks from #{path}: #{e.message}")
      logger.debug(e.backtrace.join("\n"))
    end
  end
end

#process_args(argv) ⇒ Object

Raises:



25
26
27
28
29
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
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/background_queue/server_lib/server.rb', line 25

def process_args(argv)
  argv = argv.clone
  cmd = argv.shift
  
  if cmd.nil?
    raise BackgroundQueue::ServerLib::InitError, "Usage: server command [options]"
  end
  
  options = {:command=>cmd.nil? ? nil : cmd.downcase.intern}
  
  env_to_load = "development"
  
  OptionParser.new do |opts|
    opts.banner = "Usage: server command [options]"
    case options[:command]
      when :start, :test
        opts.on("-c", "--config PATH", "Configuration Path") do |cp|
          options[:config] = cp
        end
      when :stop
        
      when nil
          
      else
        raise "Invalid Command: #{cmd}"
    end
    opts.on("-l", "--logfile [PATH]", "Logfile Path") do |lf|
      options[:log_file] = lf
    end
    opts.on("-v", "--loglevel [LEVEL]", "Log Level") do |ll|
      options[:log_level] = ll
    end
    opts.on("-p", "--pidfile [PATH]", "Pid file Path (/var/run/background_queue.pid)") do |pf|
      options[:pid_file] = pf
    end
    opts.on("-e", "--environment [RAILS_ENV]", "testing/development/production (development)") do |env|
      env_to_load = env
    end
  end.parse!(argv)
  
  ENV['RAILS_ENV']=env_to_load
  
  raise BackgroundQueue::ServerLib::InitError, "Missing config argument (-c)" if options[:config].nil? && ([:test, :start].include?(options[:command]) )

  options
end

#remove_pid(options) ⇒ Object 



190
191
192
193
194
195
 
# File 'lib/background_queue/server_lib/server.rb', line 190

def remove_pid(options)
  begin
    File.delete(get_pid_path(options))
  rescue 
  end
end

#resolve_logging_path(path) ⇒ Object 



77
78
79
 
# File 'lib/background_queue/server_lib/server.rb', line 77

def resolve_logging_path(path)
  File.expand_path(path)
end

#run(options) ⇒ Object 



245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/background_queue/server_lib/server.rb', line 245

def run(options)
  trap_signals
  @running = true
  @thread_manager = BackgroundQueue::ServerLib::ThreadManager.new(self, self.config.connections_per_worker)
  
  @workers = BackgroundQueue::ServerLib::WorkerBalancer.new(self)
  @task_queue = BackgroundQueue::ServerLib::BalancedQueue.new(self)
  
  @thread_manager.start(BackgroundQueue::ServerLib::WorkerThread)
  
  @event_server = BackgroundQueue::ServerLib::EventServer.new(self)
  
  @jobs = BackgroundQueue::ServerLib::JobRegistry.new
  
  load_tasks(config.task_file)
  
  @event_server.start
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


241
242
243
 
# File 'lib/background_queue/server_lib/server.rb', line 241

def running?
  @running
end

#save_tasks(path) ⇒ Object 



297
298
299
300
301
302
303
304
305
306
307
308
 
# File 'lib/background_queue/server_lib/server.rb', line 297

def save_tasks(path)
  return if path.nil?
  
  begin
    File.open(path, 'w') { |io| 
      task_queue.save_to_file(io)
    }
  rescue Exception=>e
    logger.error("Error saving tasks to #{path}: #{e.message}")
    logger.debug(e.backtrace.join("\n"))
  end
end

#set_logging_level(log, level) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/background_queue/server_lib/server.rb', line 81

def set_logging_level(log, level)
  if level.nil? || level.strip.length == 0
    level = "warn" 
  else
    level = level.to_s.downcase
  end
  case level
  when 'debug'
    log.level = Logger::DEBUG
  when 'info'
    log.level = Logger::INFO
  when 'warn'
    log.level = Logger::WARN
  when 'error'
    log.level = Logger::ERROR
  when 'fatal'
    log.level = Logger::FATAL
  else
    raise BackgroundQueue::ServerLib::InitError, "Unknown logging level: #{level}"
  end
end

#start(options) ⇒ Object 



220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/background_queue/server_lib/server.rb', line 220

def start(options)
  begin
    load_configuration(options[:config])
    init_logging(options[:log_file], options[:log_level])
    check_not_running(options) unless options[:skip_pid]
    write_pid(options) unless options[:skip_pid] #this will make sure we can write the pid file... the daemon will write it again
    if options[:command] == :start
      daemonize(options)
    elsif options[:command] == :run
      run(options)
    else
      raise BackgroundQueue::ServerLib::InitError, "Unknown Command: #{options[:command]}"
    end
  rescue BackgroundQueue::ServerLib::InitError=>ie
    STDERR.puts ie.message
  rescue Exception=>e
    STDERR.puts e.message
    STDERR.puts e.backtrace.join("\n")
  end
end

#stop(timeout_secs = 10) ⇒ Object 



264
265
266
267
268
269
 
# File 'lib/background_queue/server_lib/server.rb', line 264

def stop(timeout_secs=10)
  @running = false
  @event_server.stop
  @thread_manager.wait(timeout_secs)
  save_tasks(config.task_file)
end

#stop_pid(options) ⇒ Object 



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/background_queue/server_lib/server.rb', line 153

def stop_pid(options)
  proc_id = get_pid(options)
  unless proc_id.nil?
    begin
      Process.kill(15, proc_id) 
    rescue 
      #dont care... the process may have died already?
    end
    count = 0
    while get_pid(options) && count < 10
      puts "Waiting..."
      sleep(1)
    end
    kill_pid(options) #make sure
  end
end

#trap_signals ⇒ Object 



197
198
199
200
201
202
 
# File 'lib/background_queue/server_lib/server.rb', line 197

def trap_signals
  Signal.trap("TERM") do
    puts "Terminating..."
    self.stop()
  end
end

#write_pid(options) ⇒ Object 



179
180
181
182
183
184
185
186
187
188
 
# File 'lib/background_queue/server_lib/server.rb', line 179

def write_pid(options)
  proc_id = Process.pid
  begin
    File.open(get_pid_path(options), "w") { |f|
      f.write(proc_id.to_s)
    }
  rescue Exception=>e
    raise BackgroundQueue::ServerLib::InitError, "Unable to write to pid file #{get_pid_path(options)}: #{e.message}"
  end
end