waitReply and rawUnary credit the connection

bench-tool/bench-tool.cabal

@@ -1,10 +1,10 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.33.0.
+-- This file has been generated from package.yaml by hpack version 0.35.0.
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 3e8de76afd2731e91b6482e417c20c427aab6ea11675e4a6fd9acc874a760d82
+-- hash: 0c2988657eb0fb11bf40db6480adf1974e85fc19f0631ae2e22b42456a0316c5
 
 name:           bench-tool
 version:        0.4.0.1
@@ -40,16 +40,16 @@ library
       async >=2.2.1 && <3
     , base >=4.10 && <5
     , binary >=0.8.5 && <0.9
-    , bytestring >=0.10.8 && <0.11
+    , bytestring >=0.10.8 && <0.12
     , case-insensitive >=1.2.0 && <1.3
     , containers
-    , http-types >=0.12 && <0.13
-    , http2 >=1.6 && <3
+    , http-types ==0.12.*
+    , http2 >=1.6 && <3.0.4
     , http2-client
     , http2-client-grpc
     , http2-grpc-proto-lens
     , http2-grpc-proto3-wire
-    , http2-grpc-types >=0.5 && <0.6
+    , http2-grpc-types ==0.5.*
     , lifted-async
     , lifted-base
     , microlens
@@ -76,16 +76,16 @@ executable bench-tool-exe
     , base >=4.10 && <5
     , bench-tool
     , binary >=0.8.5 && <0.9
-    , bytestring >=0.10.8 && <0.11
+    , bytestring >=0.10.8 && <0.12
     , case-insensitive >=1.2.0 && <1.3
     , containers
-    , http-types >=0.12 && <0.13
-    , http2 >=1.6 && <3
+    , http-types ==0.12.*
+    , http2 >=1.6 && <3.0.4
     , http2-client
     , http2-client-grpc
     , http2-grpc-proto-lens
     , http2-grpc-proto3-wire
-    , http2-grpc-types >=0.5 && <0.6
+    , http2-grpc-types ==0.5.*
     , lifted-async
     , lifted-base
     , microlens

bench-tool/package.yaml

@@ -19,9 +19,9 @@ dependencies:
 - base >= 4.10 && < 5
 - async >= 2.2.1 && < 3
 - binary >= 0.8.5 && < 0.9
-- bytestring >= 0.10.8 && < 0.11
+- bytestring >= 0.10.8 && < 0.12
 - case-insensitive >= 1.2.0 && < 1.3
-- http2 >= 1.6 && < 3
+- http2 >= 1.6 && < 3.0.4
 - http2-grpc-types >= 0.5 && < 0.6
 - http-types >= 0.12 && < 0.13
 - unliftio-core >= 0.1 && < 0.3

bench-tool/src/BenchTool.hs

@@ -2,6 +2,7 @@
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE NumericUnderscores #-}
+{-# LANGUAGE PackageImports #-}
 module BenchTool where
 
 
@@ -18,7 +19,7 @@ import Network.GRPC.HTTP2.Encoding as Encoding
 
 -- client
 import Network.GRPC.Client (open, singleRequest, streamReply, streamRequest, Timeout(..), StreamDone(..), CompressMode(..))
-import qualified Network.HTTP2.Client as Client
+import qualified "http2-client" Network.HTTP2.Client as Client
 import qualified Network.GRPC.Client.Helpers as Client
 
 -- server

bench-tool/src/Proto/Protos/Bench.hs

@@ -173,8 +173,7 @@ instance Data.ProtoLens.Message BenchObject where
                                   (Prelude.show (missing :: [Prelude.String]))))
                       Prelude.return
                         (Lens.Family2.over
-                           Data.ProtoLens.unknownFields
-                           (\ !t -> Prelude.reverse t)
+                           Data.ProtoLens.unknownFields (\ !t -> Prelude.reverse t)
                            (Lens.Family2.set
                               (Data.ProtoLens.Field.field @"vec'strings") frozen'strings x))
                else
@@ -234,8 +233,7 @@ instance Data.ProtoLens.Message BenchObject where
                                    (Data.ProtoLens.Encoding.Bytes.putVarInt
                                       (Prelude.fromIntegral (Data.ByteString.length bs)))
                                    (Data.ProtoLens.Encoding.Bytes.putBytes bs))
-                           Data.ProtoLens.encodeMessage
-                           _v))
+                           Data.ProtoLens.encodeMessage _v))
                 (Lens.Family2.view (Data.ProtoLens.Field.field @"vec'strings") _x))
              ((Data.Monoid.<>)
                 (case
@@ -251,8 +249,7 @@ instance Data.ProtoLens.Message BenchObject where
                                      (Data.ProtoLens.Encoding.Bytes.putVarInt
                                         (Prelude.fromIntegral (Data.ByteString.length bs)))
                                      (Data.ProtoLens.Encoding.Bytes.putBytes bs))
-                             Data.ProtoLens.encodeMessage
-                             _v))
+                             Data.ProtoLens.encodeMessage _v))
                 ((Data.Monoid.<>)
                    (case
                         Lens.Family2.view (Data.ProtoLens.Field.field @"maybe'req") _x
@@ -267,8 +264,7 @@ instance Data.ProtoLens.Message BenchObject where
                                         (Data.ProtoLens.Encoding.Bytes.putVarInt
                                            (Prelude.fromIntegral (Data.ByteString.length bs)))
                                         (Data.ProtoLens.Encoding.Bytes.putBytes bs))
-                                Data.ProtoLens.encodeMessage
-                                _v))
+                                Data.ProtoLens.encodeMessage _v))
                    (Data.ProtoLens.Encoding.Wire.buildFieldSet
                       (Lens.Family2.view Data.ProtoLens.unknownFields _x))))
 instance Control.DeepSeq.NFData BenchObject where
@@ -609,8 +605,7 @@ instance Data.ProtoLens.Message BenchObject'StringContent where
                                  (Data.ProtoLens.Encoding.Bytes.putVarInt
                                     (Prelude.fromIntegral (Data.ByteString.length bs)))
                                  (Data.ProtoLens.Encoding.Bytes.putBytes bs))
-                         Data.Text.Encoding.encodeUtf8
-                         _v))
+                         Data.Text.Encoding.encodeUtf8 _v))
              (Data.ProtoLens.Encoding.Wire.buildFieldSet
                 (Lens.Family2.view Data.ProtoLens.unknownFields _x))
 instance Control.DeepSeq.NFData BenchObject'StringContent where
@@ -627,6 +622,13 @@ instance Data.ProtoLens.Service.Types.Service Bench where
   type ServiceMethods Bench = '["clientStreamQuery",
                                 "serverStreamQuery",
                                 "unaryQuery"]
+  packedServiceDescriptor _
+    = "\n\
+      \\ENQBench\DC26\n\
+      \\n\
+      \UnaryQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL\DC2?\n\
+      \\DC1ServerStreamQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL0\SOH\DC2?\n\
+      \\DC1ClientStreamQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL(\SOH"
 instance Data.ProtoLens.Service.Types.HasMethodImpl Bench "unaryQuery" where
   type MethodName Bench "unaryQuery" = "UnaryQuery"
   type MethodInput Bench "unaryQuery" = BenchObject
@@ -662,12 +664,12 @@ packedFileDescriptor
     \\n\
     \UnaryQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL\DC2?\n\
     \\DC1ServerStreamQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL0\SOH\DC2?\n\
-    \\DC1ClientStreamQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL(\SOHJ\162\ACK\n\
+    \\DC1ClientStreamQuery\DC2\DC2.bench.BenchObject\SUB\DC2.bench.BenchObject\"\NUL(\SOHJ\209\ENQ\n\
     \\ACK\DC2\EOT\NUL\NUL\ETB\SOH\n\
     \\b\n\
     \\SOH\f\DC2\ETX\NUL\NUL\DC2\n\
     \\b\n\
-    \\SOH\STX\DC2\ETX\STX\b\r\n\
+    \\SOH\STX\DC2\ETX\STX\NUL\SO\n\
     \\n\
     \\n\
     \\STX\ACK\NUL\DC2\EOT\EOT\NUL\b\SOH\n\
@@ -717,8 +719,6 @@ packedFileDescriptor
     \\ETB\n\
     \\r\n\
     \\ACK\EOT\NUL\ETX\NUL\STX\NUL\DC2\ETX\f\EOT\ETB\n\
-    \\SI\n\
-    \\a\EOT\NUL\ETX\NUL\STX\NUL\EOT\DC2\EOT\f\EOT\v\EM\n\
     \\SO\n\
     \\a\EOT\NUL\ETX\NUL\STX\NUL\ENQ\DC2\ETX\f\EOT\n\
     \\n\
@@ -733,8 +733,6 @@ packedFileDescriptor
     \\SI\n\
     \\r\n\
     \\ACK\EOT\NUL\ETX\SOH\STX\NUL\DC2\ETX\SI\EOT\DC4\n\
-    \\SI\n\
-    \\a\EOT\NUL\ETX\SOH\STX\NUL\EOT\DC2\EOT\SI\EOT\SO\DC1\n\
     \\SO\n\
     \\a\EOT\NUL\ETX\SOH\STX\NUL\ENQ\DC2\ETX\SI\EOT\t\n\
     \\SO\n\
@@ -749,8 +747,6 @@ packedFileDescriptor
     \\r\n\
     \\r\n\
     \\ACK\EOT\NUL\ETX\STX\STX\NUL\DC2\ETX\DC2\EOT\EM\n\
-    \\SI\n\
-    \\a\EOT\NUL\ETX\STX\STX\NUL\EOT\DC2\EOT\DC2\EOT\DC1\SI\n\
     \\SO\n\
     \\a\EOT\NUL\ETX\STX\STX\NUL\ENQ\DC2\ETX\DC2\EOT\t\n\
     \\SO\n\
@@ -771,8 +767,6 @@ packedFileDescriptor
     \\ENQ\EOT\NUL\STX\NUL\ETX\DC2\ETX\DC4#$\n\
     \\v\n\
     \\EOT\EOT\NUL\STX\SOH\DC2\ETX\NAK\STX\DC2\n\
-    \\r\n\
-    \\ENQ\EOT\NUL\STX\SOH\EOT\DC2\EOT\NAK\STX\DC4%\n\
     \\f\n\
     \\ENQ\EOT\NUL\STX\SOH\ACK\DC2\ETX\NAK\STX\a\n\
     \\f\n\
@@ -781,8 +775,6 @@ packedFileDescriptor
     \\ENQ\EOT\NUL\STX\SOH\ETX\DC2\ETX\NAK\DLE\DC1\n\
     \\v\n\
     \\EOT\EOT\NUL\STX\STX\DC2\ETX\SYN\STX\SO\n\
-    \\r\n\
-    \\ENQ\EOT\NUL\STX\STX\EOT\DC2\EOT\SYN\STX\NAK\DC2\n\
     \\f\n\
     \\ENQ\EOT\NUL\STX\STX\ACK\DC2\ETX\SYN\STX\ENQ\n\
     \\f\n\

http2-client-grpc/Changelog.md

@@ -2,7 +2,11 @@
 
 ## Unreleased changes
 
-Support connecting to unix sockets and already connected sockets. [#44](https://github.com/haskell-grpc-native/http2-grpc-haskell/pull/44)
+- Support connecting to unix sockets and already connected sockets. [#44](https://github.com/haskell-grpc-native/http2-grpc-haskell/pull/44)
+- Make `waitReply` not rely on the background thread. Decreases latency of `rawUnary` and related queries..
+- Fix a bug where a streaming call which failed immediately would lead to a
+  hanging request.
+
 
 ## 0.8.0.0
 

http2-client-grpc/http2-client-grpc.cabal

@@ -20,13 +20,13 @@ library
   build-depends:       base >= 4.10 && < 5
                      , async >= 2.2 && < 2.3
                      , binary >= 0.8 && < 0.9
-                     , bytestring >= 0.10.8 && < 0.11
+                     , bytestring >= 0.10.8 && < 0.11.5
                      , case-insensitive >= 1.2.0 && < 1.3
                      , data-default-class >= 0.1 && <0.2
                      , lifted-async >= 0.10 && < 0.11
                      , lifted-base >= 0.2 && < 0.3
-                     , http2 >= 1.6 && < 2.1
-                     , http2-client >= 0.10 && < 0.11
+                     , http2 >= 1.6
+                     , http2-client >= 0.10.0.1 && < 0.12
                      , http2-grpc-types >= 0.5 && < 0.6
                      , text >= 1.2 && < 1.3
                      , tls >= 1.4 && < 1.6

http2-client-grpc/src/Network/GRPC/Client.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE OverloadedStrings   #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeFamilies        #-}
+{-# LANGUAGE PackageImports      #-}
 
 -- | A module adding support for gRPC over HTTP2.
 --
@@ -94,9 +95,9 @@ import Data.Monoid ((<>))
 
 import Network.GRPC.HTTP2.Types
 import Network.GRPC.HTTP2.Encoding
-import Network.HTTP2
+import Network.HTTP2.Frame
 import Network.HPACK
-import Network.HTTP2.Client hiding (next)
+import "http2-client" Network.HTTP2.Client hiding (next)
 import Network.HTTP2.Client.Helpers
 
 type CIHeaderList = [(CI ByteString, ByteString)]
@@ -126,10 +127,10 @@ throwOnPushPromise _ _ _ _ _ = lift $ throwIO UnallowedPushPromiseReceived
 -- | Wait for an RPC reply.
 waitReply
   :: (GRPCOutput r o)
-  => r -> Decoding -> Http2Stream -> IncomingFlowControl
+  => r -> Decoding -> Http2Client -> Http2Stream -> IncomingFlowControl
   -> ClientIO (RawReply o)
-waitReply rpc decoding stream flowControl =
-    format . fromStreamResult <$> waitStream stream flowControl throwOnPushPromise
+waitReply rpc decoding conn stream flowControl =
+    format . fromStreamResult <$> waitStream conn stream flowControl throwOnPushPromise
   where
     decompress = _getDecodingCompression decoding
     format rsp = do
@@ -231,14 +232,22 @@ streamReply rpc v0 req handler = RPCCall rpc $ \conn stream isfc osfc encoding d
                 liftIO $ _addCredit isfc (ByteString.length dat)
                 _ <- liftIO $ _consumeCredit isfc (ByteString.length dat)
                 _ <- _updateWindow isfc
+                _ <- _updateWindow (_incomingFlowControl conn)
                 handleAllChunks decoding v1 hdrs decode dat loop
     } in do
         let ocfc = _outgoingFlowControl conn
         let decompress = _getDecodingCompression decoding
         sendSingleMessage rpc req encoding setEndStream conn ocfc stream osfc
         _waitEvent stream >>= \case
-            StreamHeadersEvent _ hdrs ->
-                loop v0 (decodeOutput rpc decompress) hdrs
+            StreamHeadersEvent frHeaders hdrs ->
+                   -- In case the response is an immediate error, there will typically
+                   -- not be any additional data, so we cannot enter the loop above,
+                   -- since that will not receive any additional events. So we terminate early.
+                   if testEndStream (flags frHeaders) then
+                     -- In this scenario there are only trailers, we only received that frame.
+                     return (v0, [], hdrs)
+                   else
+                     loop v0 (decodeOutput rpc decompress) hdrs
             _                         ->
                 lift $ throwIO (InvalidState "no headers")
   where
@@ -283,7 +292,7 @@ streamRequest rpc v0 handler = RPCCall  rpc$ \conn stream isfc streamFlowControl
                     go v2
                 Left _ -> do
                     sendData conn stream setEndStream ""
-                    reply <- waitReply rpc decoding stream isfc
+                    reply <- waitReply rpc decoding conn stream isfc
                     pure (v2, reply)
     in go v0
 
@@ -325,7 +334,7 @@ singleRequest
 singleRequest rpc msg = RPCCall rpc $ \conn stream isfc osfc encoding decoding -> do
     let ocfc = _outgoingFlowControl conn
     sendSingleMessage rpc msg encoding setEndStream conn ocfc stream osfc
-    waitReply rpc decoding stream isfc
+    waitReply rpc decoding conn stream isfc
 
 -- | Handler for received message.
 type HandleMessageStep i o a = HeaderList -> a -> o -> ClientIO a

http2-client-grpc/src/Network/GRPC/Client/Helpers.hs

@@ -3,6 +3,7 @@
 {-# LANGUAGE GADTs             #-}
 {-# LANGUAGE FlexibleContexts  #-}
 {-# LANGUAGE CPP               #-}
+{-# LANGUAGE PackageImports #-}
 
 -- | Set of helpers helping with writing gRPC clients with not much exposure of
 -- the http2-client complexity.
@@ -32,7 +33,7 @@ import Network.HPACK (HeaderList)
 import Data.Monoid ((<>))
 #endif
 
-import Network.HTTP2.Client (frameHttp2RawConnection, ClientIO, ClientError, newHttp2FrameConnection, newHttp2Client, Http2Client(..), IncomingFlowControl(..), GoAwayHandler, FallBackFrameHandler, ignoreFallbackHandler, HostName, PortNumber, TooMuchConcurrency)
+import "http2-client" Network.HTTP2.Client (ClientError, ClientIO, FallBackFrameHandler, GoAwayHandler, HostName, Http2Client (..), IncomingFlowControl (..), PortNumber, TooMuchConcurrency, frameHttp2RawConnection, ignoreFallbackHandler, newHttp2Client, newHttp2FrameConnection)
 import Network.HTTP2.Client.Helpers (ping)
 import Network.HTTP2.Client.RawConnection (newRawHttp2ConnectionSocket, newRawHttp2ConnectionUnix)
 import Network.GRPC.Client

http2-grpc-proto-lens/http2-grpc-proto-lens.cabal

@@ -1,10 +1,8 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.31.2.
+-- This file has been generated from package.yaml by hpack version 0.35.1.
 --
 -- see: https://github.com/sol/hpack
---
--- hash: 57a4f4c88bc0fead3bcfeb0a5ceb013176d1102a131d797f0d2273d03db1c760
 
 name:           http2-grpc-proto-lens
 version:        0.1.0.0
@@ -38,7 +36,7 @@ library
   build-depends:
       base >=4.10 && <5
     , binary >=0.8.5 && <0.11
-    , bytestring >=0.10.8 && <0.11
+    , bytestring >=0.10.8 && <0.11.5
     , case-insensitive >=1.2.0 && <1.3
     , http2-grpc-types
     , proto-lens >=0.5 && <0.8

http2-grpc-proto-lens/package.yaml

@@ -23,7 +23,7 @@ dependencies:
 - base >= 4.10 && < 5
 - http2-grpc-types
 - binary >= 0.8.5 && < 0.11
-- bytestring >= 0.10.8 && < 0.11
+- bytestring >= 0.10.8 && < 0.11.5
 - case-insensitive >= 1.2.0 && < 1.3
 - zlib >= 0.6.2 && < 0.7
 - proto-lens >= 0.5 && < 0.8

http2-grpc-proto-lens/src/Network/GRPC/HTTP2/ProtoLens.hs

@@ -16,6 +16,7 @@ import           Data.ProtoLens.Message (Message)
 import           Data.ProtoLens.Service.Types (Service(..), HasMethod, HasMethodImpl(..))
 import           Data.Proxy (Proxy(..))
 import           GHC.TypeLits (Symbol, symbolVal)
+import           Data.Kind
 
 #if MIN_VERSION_base(4,11,0)
 #else
@@ -26,7 +27,7 @@ import Network.GRPC.HTTP2.Types
 import Network.GRPC.HTTP2.Encoding
 
 -- | A proxy type for giving static information about RPCs.
-data RPC (s :: *) (m :: Symbol) = RPC
+data RPC (s :: Type) (m :: Symbol) = RPC
 
 instance (Service s, HasMethod s m) => IsRPC (RPC s m) where
   path rpc = "/" <> pkg rpc Proxy <> "." <> srv rpc Proxy <> "/" <> meth rpc Proxy