Class: GRPC::RpcServer::Pool
- Inherits:
-
Object
- Object
- GRPC::RpcServer::Pool
- Defined in:
- lib/grpc/generic/rpc_server.rb
Overview
Pool is a simple thread pool for running server requests.
Instance Method Summary collapse
-
#initialize(size) ⇒ Pool
constructor
A new instance of Pool.
-
#jobs_waiting ⇒ Object
Returns the number of jobs waiting.
-
#schedule(*args, &blk) ⇒ Object
Runs the given block on the queue with the provided args.
-
#start ⇒ Object
Starts running the jobs in the thread pool.
-
#stop ⇒ Object
Stops the jobs in the pool.
Constructor Details
#initialize(size) ⇒ Pool
Returns a new instance of Pool.
278 279 280 281 282 283 284 285 286 |
# File 'lib/grpc/generic/rpc_server.rb', line 278 def initialize(size) fail 'pool size must be positive' unless size > 0 @jobs = Queue.new @size = size @stopped = false @stop_mutex = Mutex.new @stop_cond = ConditionVariable.new @workers = [] end |
Instance Method Details
#jobs_waiting ⇒ Object
Returns the number of jobs waiting
289 290 291 |
# File 'lib/grpc/generic/rpc_server.rb', line 289 def jobs_waiting @jobs.size end |
#schedule(*args, &blk) ⇒ Object
Runs the given block on the queue with the provided args.
297 298 299 300 301 302 |
# File 'lib/grpc/generic/rpc_server.rb', line 297 def schedule(*args, &blk) fail 'already stopped' if @stopped return if blk.nil? logger.info('schedule another job') @jobs << [blk, args] end |
#start ⇒ Object
Starts running the jobs in the thread pool.
305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 |
# File 'lib/grpc/generic/rpc_server.rb', line 305 def start fail 'already stopped' if @stopped until @workers.size == @size.to_i next_thread = Thread.new do catch(:exit) do # allows { throw :exit } to kill a thread loop do begin blk, args = @jobs.pop blk.call(*args) rescue StandardError => e logger.warn('Error in worker thread') logger.warn(e) end end end # removes the threads from workers, and signal when all the # threads are complete. @stop_mutex.synchronize do @workers.delete(Thread.current) @stop_cond.signal if @workers.size == 0 end end @workers << next_thread end end |
#stop ⇒ Object
Stops the jobs in the pool
333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 |
# File 'lib/grpc/generic/rpc_server.rb', line 333 def stop logger.info('stopping, will wait for all the workers to exit') @workers.size.times { schedule { throw :exit } } @stopped = true # TODO: allow configuration of the keepalive period keep_alive = 5 @stop_mutex.synchronize do @stop_cond.wait(@stop_mutex, keep_alive) if @workers.size > 0 end # Forcibly shutdown any threads that are still alive. if @workers.size > 0 logger.warn("forcibly terminating #{@workers.size} worker(s)") @workers.each do |t| next unless t.alive? begin t.exit rescue StandardError => e logger.warn('error while terminating a worker') logger.warn(e) end end end logger.info('stopped, all workers are shutdown') end |