[ruby] experimental client side fork support (grpc#33430)

Adds experimental fork support to gRPC/Ruby

Works towards grpc#8798 (see caveats for why this wasn't marked fixed yet)
Works towards grpc#33578 (see caveats for why this wasn't marked fixed yet)

This leverages existing `pthread_atfork` based C-core support for
forking that python/php use, but there's a bit extra involved mainly
because gRPC/Ruby has additional background threads.

New tests under `src/ruby/end2end` show example usage.

Based on grpc#33495

Caveats:
- Bidi streams are not yet supported (bidi streams spawn background
threads which are not yet fork safe)
- Servers not supported
- Only linux supported

src/ruby/end2end/bad_usage_fork_test.rb

@@ -0,0 +1,92 @@
+#!/usr/bin/env ruby
+#
+# Copyright 2016 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+ENV['GRPC_ENABLE_FORK_SUPPORT'] = "1"
+fail "forking only supported on linux" unless RUBY_PLATFORM =~ /linux/
+
+this_dir = File.expand_path(File.dirname(__FILE__))
+protos_lib_dir = File.join(this_dir, 'lib')
+grpc_lib_dir = File.join(File.dirname(this_dir), 'lib')
+$LOAD_PATH.unshift(grpc_lib_dir) unless $LOAD_PATH.include?(grpc_lib_dir)
+$LOAD_PATH.unshift(protos_lib_dir) unless $LOAD_PATH.include?(protos_lib_dir)
+$LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
+
+require 'grpc'
+require 'end2end_common'
+
+def do_rpc(stub)
+  stub.echo(Echo::EchoRequest.new(request: 'hello'), deadline: Time.now + 1)
+rescue GRPC::Unavailable => e
+  STDERR.puts "RPC terminated with expected error: #{e}"
+rescue GRPC::DeadlineExceeded => e
+  STDERR.puts "RPC terminated with expected error: #{e}"
+end
+
+def expect_error_for(action)
+  STDERR.puts "#{action}: begin (pid=#{Process.pid})"
+  begin
+    yield
+  rescue RuntimeError => e
+    STDERR.puts "got (expected) error: #{e}"
+    STDERR.puts "#{action}: done (pid=#{Process.pid})"
+    return
+  end
+  fail "expected an exception due to: #{action}"
+end
+
+def main
+  # TODO(apolcyn): point this to a guaranteed-non-listening port
+  stub = Echo::EchoServer::Stub.new("localhost:443", :this_channel_is_insecure)
+  do_rpc(stub)
+  STDERR.puts "GRPC::pre_fork begin"
+  t = Thread.new do
+    expect_error_for("running prefork in a different thread than gRPC was initialized on") do
+      GRPC.prefork
+    end
+  end
+  t.join
+  expect_error_for("calling postfork_parent before prefork") { GRPC.postfork_parent }
+  expect_error_for("calling postfork_child before prefork") { GRPC.postfork_child }
+  with_logging("parent: GRPC.prefork") { GRPC.prefork }
+  expect_error_for("calling prefork twice") { GRPC.prefork }
+  expect_error_for("calling prefork twice") { GRPC.prefork }
+  expect_error_for("using gRPC after prefork") { do_rpc(stub) }
+  pid = fork do
+    expect_error_for("using gRPC before postfork_child") { do_rpc(stub) }
+    expect_error_for("calling postfork_parent from child") { GRPC.postfork_parent }
+    with_logging("child: GRPC.postfork_child") { GRPC.postfork_child }
+    expect_error_for("calling postfork_child twice") { GRPC.postfork_child }
+    with_logging("child: first post-fork RPC") { do_rpc(stub) }
+    with_logging("child: second post-fork RPC") { do_rpc(stub) }
+    STDERR.puts "child: done"
+  end
+  expect_error_for("using gRPC before postfork_parent") { do_rpc(stub) }
+  expect_error_for("calling postfork_child from parent") { GRPC.postfork_child }
+  t = Thread.new do
+    expect_error_for("running postfork_parent in a different thread than gRPC was initialized on") do
+      GRPC.postfork_parent
+    end
+  end
+  t.join
+  with_logging("parent: GRPC.postfork_parent") { GRPC.postfork_parent }
+  expect_error_for("calling postfork_parent twice") { GRPC.postfork_parent }
+  with_logging("parent: first post-fork RPC") { do_rpc(stub) }
+  with_logging("parent: second post-fork RPC") { do_rpc(stub) }
+  Process.wait pid
+  STDERR.puts "parent: done"
+end
+
+main

src/ruby/end2end/echo_server.rb

@@ -0,0 +1,60 @@
+#!/usr/bin/env ruby
+#
+# Copyright 2016 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+this_dir = File.expand_path(File.dirname(__FILE__))
+protos_lib_dir = File.join(this_dir, 'lib')
+grpc_lib_dir = File.join(File.dirname(this_dir), 'lib')
+$LOAD_PATH.unshift(grpc_lib_dir) unless $LOAD_PATH.include?(grpc_lib_dir)
+$LOAD_PATH.unshift(protos_lib_dir) unless $LOAD_PATH.include?(protos_lib_dir)
+$LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
+
+require 'grpc'
+require 'end2end_common'
+
+def create_server_creds
+  test_root = File.join(File.dirname(__FILE__), '..', 'spec', 'testdata')
+  GRPC.logger.info("test root: #{test_root}")
+  files = ['ca.pem', 'server1.key', 'server1.pem']
+  creds = files.map { |f| File.open(File.join(test_root, f)).read }
+  GRPC::Core::ServerCredentials.new(
+    creds[0],
+    [{ private_key: creds[1], cert_chain: creds[2] }],
+    true) # force client auth
+end
+
+# Runs an echo server. Once the server is running, this writes the port of the
+# server to stdout. Terminates after reading EOF on stdin.
+def main
+  secure = false
+  OptionParser.new do |opts|
+    opts.on('--secure') do
+      secure = true
+    end
+  end.parse!
+  STDERR.puts 'start server'
+  if secure
+    server_runner = ServerRunner.new(SecureEchoServerImpl)
+    server_runner.server_creds = create_server_creds
+  else
+    server_runner = ServerRunner.new(EchoServerImpl)
+  end
+  server_port = server_runner.run
+  p server_port
+  STDIN.read
+  server_runner.stop
+end
+
+main

src/ruby/end2end/end2end_common.rb

@@ -42,14 +42,21 @@ def initialize(value)
   end
 end
 
-# GreeterServer is simple server that implements the Helloworld Greeter server.
 class EchoServerImpl < Echo::EchoServer::Service
-  # say_hello implements the SayHello rpc method.
   def echo(echo_req, _)
     Echo::EchoReply.new(response: echo_req.request)
   end
 end
 
+class SecureEchoServerImpl < Echo::EchoServer::Service
+  def echo(echo_req, call)
+    unless call.metadata["authorization"] == 'test'
+      fail "expected authorization header with value: test"
+    end
+    Echo::EchoReply.new(response: echo_req.request)
+  end
+end
+
 # ServerRunner starts an "echo server" that test clients can make calls to
 class ServerRunner
   attr_accessor :server_creds
@@ -140,3 +147,9 @@ def report_controller_port_to_parent(parent_controller_port, client_controller_p
   m.port = client_controller_port.to_i
   stub.set_client_controller_port(m, deadline: Time.now + 10)
 end
+
+def with_logging(action)
+  STDERR.puts "#{action}: begin (pid=#{Process.pid})"
+  yield
+  STDERR.puts "#{action}: done (pid=#{Process.pid})"
+end

src/ruby/end2end/fork_test.rb

@@ -0,0 +1,78 @@
+#!/usr/bin/env ruby
+#
+# Copyright 2016 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+ENV['GRPC_ENABLE_FORK_SUPPORT'] = "1"
+fail "forking only supported on linux" unless RUBY_PLATFORM =~ /linux/
+
+this_dir = File.expand_path(File.dirname(__FILE__))
+protos_lib_dir = File.join(this_dir, 'lib')
+grpc_lib_dir = File.join(File.dirname(this_dir), 'lib')
+$LOAD_PATH.unshift(grpc_lib_dir) unless $LOAD_PATH.include?(grpc_lib_dir)
+$LOAD_PATH.unshift(protos_lib_dir) unless $LOAD_PATH.include?(protos_lib_dir)
+$LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
+
+require 'grpc'
+require 'end2end_common'
+
+def do_rpc(stub)
+  stub.echo(Echo::EchoRequest.new(request: 'hello'), deadline: Time.now + 300)
+end
+
+def run_client(stub)
+  do_rpc(stub)
+  with_logging("parent: GRPC.prefork") { GRPC.prefork }
+  pid = fork do
+    with_logging("child1: GRPC.postfork_child") { GRPC.postfork_child }
+    with_logging("child1: first post-fork RPC") { do_rpc(stub) }
+    with_logging("child1: GRPC.prefork") { GRPC.prefork }
+    pid2 = fork do
+      with_logging("child2: GRPC.postfork_child") { GRPC.postfork_child }
+      with_logging("child2: first post-fork RPC") { do_rpc(stub) }
+      with_logging("child2: second post-fork RPC") { do_rpc(stub) }
+      STDERR.puts "child2: done"
+    end
+    with_logging("child1: GRPC.postfork_parent") { GRPC.postfork_parent }
+    with_logging("child1: second post-fork RPC") { do_rpc(stub) }
+    Process.wait(pid2)
+    STDERR.puts "child1: done"
+  end
+  with_logging("parent: GRPC.postfork_parent") { GRPC.postfork_parent }
+  with_logging("parent: first post-fork RPC") { do_rpc(stub) }
+  with_logging("parent: second post-fork RPC") { do_rpc(stub) }
+  Process.wait pid
+  STDERR.puts "parent: done"
+end
+
+def main
+  this_dir = File.expand_path(File.dirname(__FILE__))
+  echo_server_path = File.join(this_dir, 'echo_server.rb')
+  to_child_r, _to_child_w = IO.pipe
+  to_parent_r, to_parent_w = IO.pipe
+  # Note gRPC has not yet been initialized, otherwise we would need to call prefork
+  # before spawn and postfork_parent after.
+  # TODO(apolcyn): consider redirecting server's stderr to a file
+  Process.spawn(RbConfig.ruby, echo_server_path, in: to_child_r, out: to_parent_w, err: "server_log")
+  to_child_r.close
+  to_parent_w.close
+  child_port = to_parent_r.gets.strip
+  STDERR.puts "server running on port: #{child_port}"
+  stub = Echo::EchoServer::Stub.new("localhost:#{child_port}", :this_channel_is_insecure)
+  2.times do
+    run_client(stub)
+  end
+end
+
+main

src/ruby/end2end/secure_fork_test.rb

@@ -0,0 +1,99 @@
+#!/usr/bin/env ruby
+#
+# Copyright 2016 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+ENV['GRPC_ENABLE_FORK_SUPPORT'] = "1"
+fail "forking only supported on linux" unless RUBY_PLATFORM =~ /linux/
+
+this_dir = File.expand_path(File.dirname(__FILE__))
+protos_lib_dir = File.join(this_dir, 'lib')
+grpc_lib_dir = File.join(File.dirname(this_dir), 'lib')
+$LOAD_PATH.unshift(grpc_lib_dir) unless $LOAD_PATH.include?(grpc_lib_dir)
+$LOAD_PATH.unshift(protos_lib_dir) unless $LOAD_PATH.include?(protos_lib_dir)
+$LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
+
+require 'grpc'
+require 'end2end_common'
+
+def do_rpc(stub)
+  stub.echo(Echo::EchoRequest.new(request: 'hello'), deadline: Time.now + 300)
+end
+
+def create_channel_creds
+  test_root = File.join(File.dirname(__FILE__), '..', 'spec', 'testdata')
+  files = ['ca.pem', 'client.key', 'client.pem']
+  creds = files.map { |f| File.open(File.join(test_root, f)).read }
+  GRPC::Core::ChannelCredentials.new(creds[0], creds[1], creds[2])
+end
+
+def client_cert
+  test_root = File.join(File.dirname(__FILE__), '..', 'spec', 'testdata')
+  cert = File.open(File.join(test_root, 'client.pem')).read
+  fail unless cert.is_a?(String)
+  cert
+end
+
+def run_client(stub)
+  do_rpc(stub)
+  with_logging("parent: GRPC.prefork") { GRPC.prefork }
+  pid = fork do
+    with_logging("child1: GRPC.postfork_child") { GRPC.postfork_child }
+    with_logging("child1: first post-fork RPC") { do_rpc(stub) }
+    with_logging("child1: GRPC.prefork") { GRPC.prefork }
+    pid2 = fork do
+      with_logging("child2: GRPC.postfork_child") { GRPC.postfork_child }
+      with_logging("child2: first post-fork RPC") { do_rpc(stub) }
+      with_logging("child2: second post-fork RPC") { do_rpc(stub) }
+      STDERR.puts "child2: done"
+    end
+    with_logging("child1: GRPC.postfork_parent") { GRPC.postfork_parent }
+    with_logging("child1: second post-fork RPC") { do_rpc(stub) }
+    Process.wait(pid2)
+    STDERR.puts "child1: done"
+  end
+  with_logging("parent: GRPC.postfork_parent") { GRPC.postfork_parent }
+  with_logging("parent: first post-fork RPC") { do_rpc(stub) }
+  with_logging("parent: second post-fork RPC") { do_rpc(stub) }
+  Process.wait pid
+  STDERR.puts "parent: done"
+end
+
+# This is designed to test fork support around three key things:
+# - call credentials (relies on a special background thread)
+# - channel credentials (ownership semantics are interesting for re-creating channels)
+# - channel args (ownership semantics are interesting for re-creating channels)
+def main
+  this_dir = File.expand_path(File.dirname(__FILE__))
+  echo_server_path = File.join(this_dir, 'echo_server.rb')
+  to_child_r, _to_child_w = IO.pipe
+  to_parent_r, to_parent_w = IO.pipe
+  Process.spawn(RbConfig.ruby, echo_server_path, "--secure", in: to_child_r, out: to_parent_w, err: "server_log")
+  to_child_r.close
+  to_parent_w.close
+  child_port = to_parent_r.gets.strip
+  STDERR.puts "server running on port: #{child_port}"
+  channel_creds = create_channel_creds.compose(
+    GRPC::Core::CallCredentials.new(proc do |args|
+      { 'authorization' => 'test' }.merge(args)
+    end))
+  stub = Echo::EchoServer::Stub.new(
+    "localhost:#{child_port}", channel_creds,
+    channel_args: { GRPC::Core::Channel::SSL_TARGET => 'foo.test.google.fr' })
+  2.times do
+    run_client(stub)
+  end
+end
+
+main