Repository: elixir-grpc/grpc
Pull request #151 (2020-03-21 13:41:12)

fix(server): remove blocking read_body by receiving in our handler
fixes https://github.com/elixir-grpc/grpc/issues/149


Patch:
diff --git a/lib/grpc/adapter/cowboy/handler.ex b/lib/grpc/adapter/cowboy/handler.ex
index f281f84..c738941 100644
--- a/lib/grpc/adapter/cowboy/handler.ex
+++ b/lib/grpc/adapter/cowboy/handler.ex
@@ -14,7 +14,8 @@ defmodule GRPC.Adapter.Cowboy.Handler do
           pid: pid,
           handling_timer: reference | nil,
           resp_trailers: map,
-          compressor: atom | nil
+          compressor: atom | nil,
+          pending_reader: nil
         }
 
   @spec init(map, {atom, GRPC.Server.servers_map(), map}) :: {:cowboy_loop, map, map}
@@ -51,7 +52,7 @@ defmodule GRPC.Adapter.Cowboy.Handler do
           )
         end
 
-      {:cowboy_loop, req, %{pid: pid, handling_timer: timer_ref}}
+      {:cowboy_loop, req, %{pid: pid, handling_timer: timer_ref, pending_reader: nil}}
     else
       {:error, error} ->
         trailers = HTTP2.server_trailers(error.status, error.message)
@@ -172,13 +173,38 @@ defmodule GRPC.Adapter.Cowboy.Handler do
 
   def info({:read_body, ref, pid}, req, state) do
     opts = timeout_left_opt(state[:handling_timer])
-    {s, body, req} = :cowboy_req.read_body(req, opts)
-    send(pid, {ref, {s, body}})
-    {:ok, req, state}
-  catch
-    :exit, :timeout ->
-      Logger.warn("Timeout when reading body")
-      info({:handling_timeout, self()}, req, state)
+
+    case async_read_body(req, opts) do
+      {:send, {s, body, req}} ->
+        send(pid, {ref, {s, body}})
+        {:ok, req, state}
+
+      {:wait, read_ref} ->
+        {:ok, req, %{state | pending_reader: {read_ref, pid, ref}}}
+    end
+  end
+
+  def info({:request_body, ref, :nofin, body}, req, %{pending_reader: {ref, pid, reader_ref}} = s) do
+    send(pid, {reader_ref, {:more, body}})
+    {:ok, req, %{s | pending_reader: nil}}
+  end
+
+  def info(
+        {:request_body, ref, :fin, body_length, body},
+        %{headers: headers} = req,
+        %{pending_reader: {ref, pid, reader_ref}} = s
+      ) do
+    send(pid, {reader_ref, {:ok, body}})
+
+    # cowboy_req's set_body_length
+    req =
+      Map.merge(req, %{
+        headers: Map.put(headers, "content-length", Integer.to_string(body_length)),
+        body_length: body_length,
+        has_read_body: true
+      })
+
+    {:ok, req, s}
   end
 
   def info({:get_headers, ref, pid}, req, state) do
@@ -433,4 +459,24 @@ defmodule GRPC.Adapter.Cowboy.Handler do
     exit_handler(pid, :rpc_error)
     send_error_trailers(req, trailers)
   end
+
+  # Similar with cowboy's read_body, but we need to receive the message
+  # in `info` callback.
+  defp async_read_body(%{has_body: false} = req, _opts) do
+    {:send, {:ok, <<>>, req}}
+  end
+
+  defp async_read_body(%{has_read_body: true} = req, _opts) do
+    {:send, {:ok, <<>>, req}}
+  end
+
+  defp async_read_body(%{pid: pid, streamid: stream_id} = req, opts) do
+    length = Map.get(opts, :length, 8_000_000)
+    period = Map.get(opts, :period, 15000)
+    ref = make_ref()
+
+    :cowboy_req.cast({:read_body, self(), ref, length, period}, req)
+
+    {:wait, ref}
+  end
 end


Test patch:
diff --git a/test/grpc/integration/service_test.exs b/test/grpc/integration/service_test.exs
index f74c94f..3da896f 100644
--- a/test/grpc/integration/service_test.exs
+++ b/test/grpc/integration/service_test.exs
@@ -39,12 +39,28 @@ defmodule GRPC.Integration.ServiceTest do
       end)
     end
 
+    def async_route_chat(req_enum, stream) do
+      tasks =
+        Enum.map(req_enum, fn note ->
+          note = %{note | message: "Reply: #{note.message}"}
+          # Send out reply asynchronosly.
+          # this makes it very likely to read client stream frist before sending out
+          Task.async(fn ->
+            Server.send_reply(stream, note)
+          end)
+        end)
+
+      Enum.each(tasks, fn task ->
+        Task.await(task)
+      end)
+    end
+
     defp simple_feature(point) do
       Routeguide.Feature.new(location: point, name: "#{point.latitude},#{point.longitude}")
     end
   end
 
-  test "Unary RPC works" do
+  test "unary RPC works" do
     run_server(FeatureServer, fn port ->
       {:ok, channel} = GRPC.Stub.connect("localhost:#{port}")
       point = Routeguide.Point.new(latitude: 409_146_138, longitude: -746_188_906)
@@ -53,7 +69,7 @@ defmodule GRPC.Integration.ServiceTest do
     end)
   end
 
-  test "Server streaming RPC works" do
+  test "server streaming RPC works" do
     run_server(FeatureServer, fn port ->
       {:ok, channel} = GRPC.Stub.connect("localhost:#{port}")
       low = Routeguide.Point.new(latitude: 400_000_000, longitude: -750_000_000)
@@ -68,7 +84,7 @@ defmodule GRPC.Integration.ServiceTest do
     end)
   end
 
-  test "Client streaming RPC works" do
+  test "client streaming RPC works" do
     run_server(FeatureServer, fn port ->
       {:ok, channel} = GRPC.Stub.connect("localhost:#{port}")
       point1 = Routeguide.Point.new(latitude: 400_000_000, longitude: -750_000_000)
@@ -81,7 +97,7 @@ defmodule GRPC.Integration.ServiceTest do
     end)
   end
 
-  test "Bidirectional streaming RPC works" do
+  test "bidirectional streaming RPC works" do
     run_server(FeatureServer, fn port ->
       {:ok, channel} = GRPC.Stub.connect("localhost:#{port}")
       stream = channel |> Routeguide.RouteGuide.Stub.route_chat()
@@ -108,4 +124,47 @@ defmodule GRPC.Integration.ServiceTest do
       assert length(notes) == 6
     end)
   end
+
+  # There was a bug that blocks reading messages while sending the replies.
+  # This is for the case.
+  test "async bidirectional streaming RPC works" do
+    run_server(
+      FeatureServer,
+      fn port ->
+        {:ok, channel} = GRPC.Stub.connect("localhost:#{port}")
+        stream = channel |> Routeguide.RouteGuide.Stub.async_route_chat()
+
+        task =
+          Task.async(fn ->
+            Enum.each(1..5, fn i ->
+              point = Routeguide.Point.new(latitude: 0, longitude: rem(i, 3) + 1)
+              note = Routeguide.RouteNote.new(location: point, message: "Message #{i}")
+              # note that we don't send end of stream yet here
+              GRPC.Stub.send_request(stream, note, [])
+            end)
+          end)
+
+        result = GRPC.Stub.recv(stream)
+
+        {:ok, result_enum} = result
+
+        Task.await(task)
+
+        notes =
+          Enum.map(result_enum, fn {:ok, note} ->
+            assert "Reply: " <> msg = note.message
+
+            if note.message == "Reply: Message 5" do
+              point = Routeguide.Point.new(latitude: 0, longitude: rem(6, 3) + 1)
+              note = Routeguide.RouteNote.new(location: point, message: "Message #{6}")
+              GRPC.Stub.send_request(stream, note, end_stream: true)
+            end
+
+            note
+          end)
+
+        assert length(notes) == 6
+      end
+    )
+  end
 end
diff --git a/test/support/route_guide.pb.ex b/test/support/route_guide.pb.ex
index f4ce674..6ad6f55 100644
--- a/test/support/route_guide.pb.ex
+++ b/test/support/route_guide.pb.ex
@@ -74,6 +74,7 @@ defmodule Routeguide.RouteGuide.Service do
   rpc :ListFeatures, Routeguide.Rectangle, stream(Routeguide.Feature)
   rpc :RecordRoute, stream(Routeguide.Point), Routeguide.RouteSummary
   rpc :RouteChat, stream(Routeguide.RouteNote), stream(Routeguide.RouteNote)
+  rpc :AsyncRouteChat, stream(Routeguide.RouteNote), stream(Routeguide.RouteNote)
 end
 
 defmodule Routeguide.RouteGuide.Stub do