Class: GRPC::RpcServer

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Core::CallOps, Core::TimeConsts
Defined in:
src/ruby/lib/grpc/generic/rpc_server.rb

Overview

RpcServer hosts a number of services and makes them available on the network.

Constant Summary collapse

DEFAULT_POOL_SIZE =

Default thread pool size is 3

3
DEFAULT_MAX_WAITING_REQUESTS =

Default max_waiting_requests size is 20

20
DEFAULT_POLL_PERIOD =

Default poll period is 1s

1
SIGNAL_CHECK_PERIOD =

Signal check period is 0.25s

0.25

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Core::TimeConsts

from_relative_time

Constructor Details

#initialize(pool_size: DEFAULT_POOL_SIZE, max_waiting_requests: DEFAULT_MAX_WAITING_REQUESTS, poll_period: DEFAULT_POLL_PERIOD, completion_queue_override: nil, server_override: nil, connect_md_proc: nil, **kw) ⇒ RpcServer

Creates a new RpcServer.

The RPC server is configured using keyword arguments.

There are some specific keyword args used to configure the RpcServer instance, however other arbitrary are allowed and when present are used to configure the listeninng connection set up by the RpcServer.

  • server_override: which if passed must be a [GRPC::Core::Server]. When

present.

  • poll_period: when present, the server polls for new events with this

period

  • pool_size: the size of the thread pool the server uses to run its

threads

  • completion_queue_override: when supplied, this will be used as the

completion_queue that the server uses to receive network events, otherwise its creates a new instance itself

  • creds: [GRPC::Core::ServerCredentials]

the credentials used to secure the server

  • max_waiting_requests: the maximum number of requests that are not

being handled to allow. When this limit is exceeded, the server responds with not available to new requests

  • connect_md_proc:

when non-nil is a proc for determining metadata to to send back the client on receiving an invocation req. The proc signature is: val, .. func(method_name, val, …)


252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 252

def initialize(pool_size:DEFAULT_POOL_SIZE,
               max_waiting_requests:DEFAULT_MAX_WAITING_REQUESTS,
               poll_period:DEFAULT_POLL_PERIOD,
               completion_queue_override:nil,
               server_override:nil,
               connect_md_proc:nil,
               **kw)
  @connect_md_proc = RpcServer.setup_connect_md_proc(connect_md_proc)
  @cq = RpcServer.setup_cq(completion_queue_override)
  @max_waiting_requests = max_waiting_requests
  @poll_period = poll_period
  @pool_size = pool_size
  @pool = Pool.new(@pool_size)
  @run_cond = ConditionVariable.new
  @run_mutex = Mutex.new
  @running = false
  @server = RpcServer.setup_srv(server_override, @cq, **kw)
  @stopped = false
  @stop_mutex = Mutex.new
end

Class Method Details

.setup_connect_md_proc(a_proc) ⇒ Object

setup_connect_md_proc is used by #initialize to validate the connect_md_proc.


214
215
216
217
218
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 214

def self.setup_connect_md_proc(a_proc)
  return nil if a_proc.nil?
  fail(TypeError, '!Proc') unless a_proc.is_a? Proc
  a_proc
end

.setup_cq(alt_cq) ⇒ Object

setup_cq is used by #initialize to constuct a Core::CompletionQueue from its arguments.


196
197
198
199
200
201
202
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 196

def self.setup_cq(alt_cq)
  return Core::CompletionQueue.new if alt_cq.nil?
  unless alt_cq.is_a? Core::CompletionQueue
    fail(TypeError, '!CompletionQueue')
  end
  alt_cq
end

.setup_srv(alt_srv, cq, **kw) ⇒ Object

setup_srv is used by #initialize to constuct a Core::Server from its arguments.


206
207
208
209
210
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 206

def self.setup_srv(alt_srv, cq, **kw)
  return Core::Server.new(cq, kw) if alt_srv.nil?
  fail(TypeError, '!Server') unless alt_srv.is_a? Core::Server
  alt_srv
end

Instance Method Details

#available?(an_rpc) ⇒ Boolean

Sends UNAVAILABLE if there are too many unprocessed jobs

Returns:

  • (Boolean)

392
393
394
395
396
397
398
399
400
401
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 392

def available?(an_rpc)
  jobs_count, max = @pool.jobs_waiting, @max_waiting_requests
  GRPC.logger.info("waiting: #{jobs_count}, max: #{max}")
  return an_rpc if @pool.jobs_waiting <= @max_waiting_requests
  GRPC.logger.warn("NOT AVAILABLE: too many jobs_waiting: #{an_rpc}")
  noop = proc { |x| x }
  c = ActiveCall.new(an_rpc.call, @cq, noop, noop, an_rpc.deadline)
  c.send_status(StatusCodes::UNAVAILABLE, '')
  nil
end

#handle(service) ⇒ Object

handle registration of classes

service is either a class that includes GRPC::GenericService and whose #new function can be called without argument or any instance of such a class.

E.g, after

class Divider

include GRPC::GenericService
rpc :div DivArgs, DivReply    # single request, single response
def initialize(optional_arg='default option') # no args
  ...
end

srv = GRPC::RpcServer.new(…)

# Either of these works

srv.handle(Divider)

# or

srv.handle(Divider.new('replace optional arg'))

It raises RuntimeError:

  • if service is not valid service class or object

  • its handler methods are already registered

  • if the server is already running

Parameters:

  • service (Object|Class)

    a service class or object as described above


362
363
364
365
366
367
368
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 362

def handle(service)
  fail 'cannot add services if the server is running' if running?
  fail 'cannot add services if the server is stopped' if stopped?
  cls = service.is_a?(Class) ? service : service.class
  assert_valid_service_class(cls)
  add_rpc_descs_for(service)
end

#implemented?(an_rpc) ⇒ Boolean

Sends UNIMPLEMENTED if the method is not implemented by this server

Returns:

  • (Boolean)

404
405
406
407
408
409
410
411
412
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 404

def implemented?(an_rpc)
  mth = an_rpc.method.to_sym
  return an_rpc if rpc_descs.key?(mth)
  GRPC.logger.warn("UNIMPLEMENTED: #{an_rpc}")
  noop = proc { |x| x }
  c = ActiveCall.new(an_rpc.call, @cq, noop, noop, an_rpc.deadline)
  c.send_status(StatusCodes::UNIMPLEMENTED, '')
  nil
end

#loop_handle_server_callsObject

handles calls to the server


415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 415

def loop_handle_server_calls
  fail 'not running' unless @running
  loop_tag = Object.new
  until stopped?
    begin
      an_rpc = @server.request_call(@cq, loop_tag, INFINITE_FUTURE)
      break if (!an_rpc.nil?) && an_rpc.call.nil?

      active_call = new_active_server_call(an_rpc)
      unless active_call.nil?
        @pool.schedule(active_call) do |ac|
          c, mth = ac
          rpc_descs[mth].run_server_method(c, rpc_handlers[mth])
        end
      end
    rescue Core::CallError, RuntimeError => e
      # these might happen for various reasonse.  The correct behaviour of
      # the server is to log them and continue, if it's not shutting down.
      GRPC.logger.warn("server call failed: #{e}") unless stopped?
      next
    end
  end
  @running = false
  GRPC.logger.info("stopped: #{self}")
end

#new_active_server_call(an_rpc) ⇒ Object


441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 441

def new_active_server_call(an_rpc)
  return nil if an_rpc.nil? || an_rpc.call.nil?

  # allow the metadata to be accessed from the call
  handle_call_tag = Object.new
  an_rpc.call. = an_rpc.  # attaches md to call for handlers
  GRPC.logger.debug("call md is #{an_rpc.metadata}")
  connect_md = nil
  unless @connect_md_proc.nil?
    connect_md = @connect_md_proc.call(an_rpc.method, an_rpc.)
  end
  an_rpc.call.run_batch(@cq, handle_call_tag, INFINITE_FUTURE,
                         => connect_md)
  return nil unless available?(an_rpc)
  return nil unless implemented?(an_rpc)

  # Create the ActiveCall
  GRPC.logger.info("deadline is #{an_rpc.deadline}; (now=#{Time.now})")
  rpc_desc = rpc_descs[an_rpc.method.to_sym]
  c = ActiveCall.new(an_rpc.call, @cq,
                     rpc_desc.marshal_proc, rpc_desc.unmarshal_proc(:input),
                     an_rpc.deadline)
  mth = an_rpc.method.to_sym
  [c, mth]
end

#runObject

runs the server

  • if no rpc_descs are registered, this exits immediately, otherwise it continues running permanently and does not return until program exit.

  • #running? returns true after this is called, until #stop cause the the server to stop.


377
378
379
380
381
382
383
384
385
386
387
388
389
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 377

def run
  if rpc_descs.size.zero?
    GRPC.logger.warn('did not run as no services were present')
    return
  end
  @run_mutex.synchronize do
    @running = true
    @run_cond.signal
  end
  @pool.start
  @server.start
  loop_handle_server_calls
end

#run_till_terminatedObject

Runs the server in its own thread, then waits for signal INT or TERM on the current thread to terminate it.


318
319
320
321
322
323
324
325
326
327
328
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 318

def run_till_terminated
  GRPC.trap_signals
  t = Thread.new { run }
  wait_till_running
  loop do
    sleep SIGNAL_CHECK_PERIOD
    break unless GRPC.handle_signals
  end
  stop
  t.join
end

#running?Boolean

determines if the server is currently running

Returns:

  • (Boolean)

297
298
299
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 297

def running?
  @running
end

#stopObject

stops a running server

the call has no impact if the server is already stopped, otherwise server's current call loop is it's last.


277
278
279
280
281
282
283
284
285
286
287
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 277

def stop
  return unless @running
  @stop_mutex.synchronize do
    @stopped = true
  end
  deadline = from_relative_time(@poll_period)
  return if @server.close(@cq, deadline)
  deadline = from_relative_time(@poll_period)
  @server.close(@cq, deadline)
  @pool.stop
end

#stopped?Boolean

determines if the server has been stopped

Returns:

  • (Boolean)

290
291
292
293
294
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 290

def stopped?
  @stop_mutex.synchronize do
    return @stopped
  end
end

#wait_till_running(timeout = 0.1) ⇒ Object

Is called from other threads to wait for #run to start up the server.

If run has not been called, this returns immediately.

Parameters:

  • timeout (Numeric) (defaults to: 0.1)

    number of seconds to wait


307
308
309
310
311
312
313
314
# File 'src/ruby/lib/grpc/generic/rpc_server.rb', line 307

def wait_till_running(timeout = 0.1)
  end_time, sleep_period = Time.now + timeout, (1.0 * timeout) / 100
  while Time.now < end_time
    @run_mutex.synchronize { @run_cond.wait(@run_mutex) } unless running?
    sleep(sleep_period)
  end
  running?
end