Ensure that there's always a poller for completion

The C core uses completion_queue_next as a directive that progress should be made on requests bound to that cq.
If nobody is polling for completion, then a deadlock can occur (client needs to do work, but we're blocked waiting for the server).
There is no scope for this to occur using the idiomatic layers (as far as I can tell), but these low level tests need to be massaged.
pull/1888/head
Craig Tiller 10 years ago
parent 0596621ec2
commit 09600be0a5
  1. 60
      src/ruby/spec/client_server_spec.rb

@ -74,6 +74,12 @@ shared_examples 'basic GRPC message delivery is OK' do
it 'servers receive requests from clients and can respond' do it 'servers receive requests from clients and can respond' do
call = new_client_call call = new_client_call
server_call = nil
server_thread = Thread.new do
server_call = server_allows_client_to_proceed
end
client_ops = { client_ops = {
CallOps::SEND_INITIAL_METADATA => {}, CallOps::SEND_INITIAL_METADATA => {},
CallOps::SEND_MESSAGE => sent_message CallOps::SEND_MESSAGE => sent_message
@ -84,7 +90,7 @@ shared_examples 'basic GRPC message delivery is OK' do
expect(batch_result.send_message).to be true expect(batch_result.send_message).to be true
# confirm the server can read the inbound message # confirm the server can read the inbound message
server_call = server_allows_client_to_proceed server_thread.join
server_ops = { server_ops = {
CallOps::RECV_MESSAGE => nil CallOps::RECV_MESSAGE => nil
} }
@ -95,6 +101,12 @@ shared_examples 'basic GRPC message delivery is OK' do
it 'responses written by servers are received by the client' do it 'responses written by servers are received by the client' do
call = new_client_call call = new_client_call
server_call = nil
server_thread = Thread.new do
server_call = server_allows_client_to_proceed
end
client_ops = { client_ops = {
CallOps::SEND_INITIAL_METADATA => {}, CallOps::SEND_INITIAL_METADATA => {},
CallOps::SEND_MESSAGE => sent_message CallOps::SEND_MESSAGE => sent_message
@ -105,7 +117,7 @@ shared_examples 'basic GRPC message delivery is OK' do
expect(batch_result.send_message).to be true expect(batch_result.send_message).to be true
# confirm the server can read the inbound message # confirm the server can read the inbound message
server_call = server_allows_client_to_proceed server_thread.join
server_ops = { server_ops = {
CallOps::RECV_MESSAGE => nil, CallOps::RECV_MESSAGE => nil,
CallOps::SEND_MESSAGE => reply_text CallOps::SEND_MESSAGE => reply_text
@ -118,6 +130,12 @@ shared_examples 'basic GRPC message delivery is OK' do
it 'servers can ignore a client write and send a status' do it 'servers can ignore a client write and send a status' do
call = new_client_call call = new_client_call
server_call = nil
server_thread = Thread.new do
server_call = server_allows_client_to_proceed
end
client_ops = { client_ops = {
CallOps::SEND_INITIAL_METADATA => {}, CallOps::SEND_INITIAL_METADATA => {},
CallOps::SEND_MESSAGE => sent_message CallOps::SEND_MESSAGE => sent_message
@ -129,7 +147,7 @@ shared_examples 'basic GRPC message delivery is OK' do
# confirm the server can read the inbound message # confirm the server can read the inbound message
the_status = Struct::Status.new(StatusCodes::OK, 'OK') the_status = Struct::Status.new(StatusCodes::OK, 'OK')
server_call = server_allows_client_to_proceed server_thread.join
server_ops = { server_ops = {
CallOps::SEND_STATUS_FROM_SERVER => the_status CallOps::SEND_STATUS_FROM_SERVER => the_status
} }
@ -141,6 +159,12 @@ shared_examples 'basic GRPC message delivery is OK' do
it 'completes calls by sending status to client and server' do it 'completes calls by sending status to client and server' do
call = new_client_call call = new_client_call
server_call = nil
server_thread = Thread.new do
server_call = server_allows_client_to_proceed
end
client_ops = { client_ops = {
CallOps::SEND_INITIAL_METADATA => {}, CallOps::SEND_INITIAL_METADATA => {},
CallOps::SEND_MESSAGE => sent_message CallOps::SEND_MESSAGE => sent_message
@ -152,7 +176,7 @@ shared_examples 'basic GRPC message delivery is OK' do
# confirm the server can read the inbound message and respond # confirm the server can read the inbound message and respond
the_status = Struct::Status.new(StatusCodes::OK, 'OK', {}) the_status = Struct::Status.new(StatusCodes::OK, 'OK', {})
server_call = server_allows_client_to_proceed server_thread.join
server_ops = { server_ops = {
CallOps::RECV_MESSAGE => nil, CallOps::RECV_MESSAGE => nil,
CallOps::SEND_MESSAGE => reply_text, CallOps::SEND_MESSAGE => reply_text,
@ -221,6 +245,11 @@ shared_examples 'GRPC metadata delivery works OK' do
it 'sends all the metadata pairs when keys and values are valid' do it 'sends all the metadata pairs when keys and values are valid' do
@valid_metadata.each do |md| @valid_metadata.each do |md|
recvd_rpc = nil
rcv_thread = Thread.new do
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
end
call = new_client_call call = new_client_call
client_ops = { client_ops = {
CallOps::SEND_INITIAL_METADATA => md CallOps::SEND_INITIAL_METADATA => md
@ -230,7 +259,7 @@ shared_examples 'GRPC metadata delivery works OK' do
expect(batch_result.send_metadata).to be true expect(batch_result.send_metadata).to be true
# confirm the server can receive the client metadata # confirm the server can receive the client metadata
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline) rcv_thread.join
expect(recvd_rpc).to_not eq nil expect(recvd_rpc).to_not eq nil
recvd_md = recvd_rpc.metadata recvd_md = recvd_rpc.metadata
replace_symbols = Hash[md.each_pair.collect { |x, y| [x.to_s, y] }] replace_symbols = Hash[md.each_pair.collect { |x, y| [x.to_s, y] }]
@ -257,6 +286,11 @@ shared_examples 'GRPC metadata delivery works OK' do
it 'raises an exception if a metadata key is invalid' do it 'raises an exception if a metadata key is invalid' do
@bad_keys.each do |md| @bad_keys.each do |md|
recvd_rpc = nil
rcv_thread = Thread.new do
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
end
call = new_client_call call = new_client_call
# client signals that it's done sending metadata to allow server to # client signals that it's done sending metadata to allow server to
# respond # respond
@ -266,7 +300,7 @@ shared_examples 'GRPC metadata delivery works OK' do
call.run_batch(@client_queue, @client_tag, deadline, client_ops) call.run_batch(@client_queue, @client_tag, deadline, client_ops)
# server gets the invocation # server gets the invocation
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline) rcv_thread.join
expect(recvd_rpc).to_not eq nil expect(recvd_rpc).to_not eq nil
server_ops = { server_ops = {
CallOps::SEND_INITIAL_METADATA => md CallOps::SEND_INITIAL_METADATA => md
@ -280,6 +314,11 @@ shared_examples 'GRPC metadata delivery works OK' do
end end
it 'sends an empty hash if no metadata is added' do it 'sends an empty hash if no metadata is added' do
recvd_rpc = nil
rcv_thread = Thread.new do
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
end
call = new_client_call call = new_client_call
# client signals that it's done sending metadata to allow server to # client signals that it's done sending metadata to allow server to
# respond # respond
@ -289,7 +328,7 @@ shared_examples 'GRPC metadata delivery works OK' do
call.run_batch(@client_queue, @client_tag, deadline, client_ops) call.run_batch(@client_queue, @client_tag, deadline, client_ops)
# server gets the invocation but sends no metadata back # server gets the invocation but sends no metadata back
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline) rcv_thread.join
expect(recvd_rpc).to_not eq nil expect(recvd_rpc).to_not eq nil
server_call = recvd_rpc.call server_call = recvd_rpc.call
server_ops = { server_ops = {
@ -308,6 +347,11 @@ shared_examples 'GRPC metadata delivery works OK' do
it 'sends all the pairs when keys and values are valid' do it 'sends all the pairs when keys and values are valid' do
@valid_metadata.each do |md| @valid_metadata.each do |md|
recvd_rpc = nil
rcv_thread = Thread.new do
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
end
call = new_client_call call = new_client_call
# client signals that it's done sending metadata to allow server to # client signals that it's done sending metadata to allow server to
# respond # respond
@ -317,7 +361,7 @@ shared_examples 'GRPC metadata delivery works OK' do
call.run_batch(@client_queue, @client_tag, deadline, client_ops) call.run_batch(@client_queue, @client_tag, deadline, client_ops)
# server gets the invocation but sends no metadata back # server gets the invocation but sends no metadata back
recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline) rcv_thread.join
expect(recvd_rpc).to_not eq nil expect(recvd_rpc).to_not eq nil
server_call = recvd_rpc.call server_call = recvd_rpc.call
server_ops = { server_ops = {

Loading…
Cancel
Save