Add tests, fix some Valgrind errors (#25)

* add tests for server misbehavior

* fix leaks in pluck and next

* fix deadline leak

* fix slice leak

* always free metadata

* add full TODO in case we get rid of unregistered code

* shorter deadline to make test faster

* update for grpc 0.14.0

* fix another valgrind complaint

cbits/grpc_haskell.c

@@ -9,7 +9,7 @@
 
 void grpc_haskell_free(char *debugMsg, void *ptr){
   #ifdef GRPC_HASKELL_DEBUG
-  printf("C wrapper: freeing %s, ptr: %p\n", debugMsg, ptr);
+  printf("C wrapper: %s: freeing ptr: %p\n", debugMsg, ptr);
   #endif
   free(ptr);
 }
@@ -175,13 +175,14 @@ grpc_op* op_array_create(size_t n){
 }
 
 void op_array_destroy(grpc_op* op_array, size_t n){
+  #ifdef GRPC_HASKELL_DEBUG
+  printf("C wrapper: entered op_array_destroy\n");
+  #endif
   for(int i = 0; i < n; i++){
     grpc_op* op = op_array + i;
     switch (op->op) {
       case GRPC_OP_SEND_INITIAL_METADATA:
-      if(op->data.send_initial_metadata.count > 0){
+      metadata_free(op->data.send_initial_metadata.metadata);
-        metadata_free(op->data.send_initial_metadata.metadata);
-      }
       break;
       case GRPC_OP_SEND_MESSAGE:
       grpc_byte_buffer_destroy(op->data.send_message);
@@ -224,6 +225,7 @@ void op_send_initial_metadata_empty(grpc_op *op_array, size_t i){
   grpc_op *op = op_array + i;
   op->op = GRPC_OP_SEND_INITIAL_METADATA;
   op->data.send_initial_metadata.count = 0;
+  op->data.send_initial_metadata.metadata = malloc(0*sizeof(grpc_metadata));
   op->flags = 0;
   op->reserved = NULL;
 }

src/Network/GRPC/LowLevel/CompletionQueue.hs

@@ -37,6 +37,7 @@ import           Control.Exception                       (bracket)
 import           Data.IORef                              (newIORef)
 import           Data.List                               (intersperse)
 import           Foreign.Marshal.Alloc                   (free, malloc)
+import           Foreign.Ptr                             (nullPtr)
 import           Foreign.Storable                        (peek)
 import qualified Network.GRPC.Unsafe                     as C
 import qualified Network.GRPC.Unsafe.Constants           as C
@@ -97,8 +98,8 @@ shutdownCompletionQueue (CompletionQueue{..}) = do
 
   where drainLoop :: IO ()
         drainLoop = do
-          deadline <- C.secondsToDeadline 1
+          ev <- C.withDeadlineSeconds 1 $ \deadline ->
-          ev <- C.grpcCompletionQueueNext unsafeCQ deadline C.reserved
+                  C.grpcCompletionQueueNext unsafeCQ deadline C.reserved
           case (C.eventCompletionType ev) of
             C.QueueShutdown -> return ()
             C.QueueTimeout -> drainLoop
@@ -170,7 +171,12 @@ serverRequestCall
                      let assembledCall = ServerCall rawCall metadataArrayPtr
                                            bbPtr Nothing deadline
                      return $ Right assembledCall
-      -- TODO: see TODO for failureCleanup in serverRequestCall.
+      --TODO: the gRPC library appears to hold onto these pointers for a random
+      -- amount of time, even after returning from the only call that uses them.
+      -- This results in malloc errors if
+      -- gRPC tries to modify them after we free them. To work around it,
+      -- we sleep for a while before freeing the objects. We should find a
+      -- permanent solution that's more robust.
       where failureCleanup deadline callPtr metadataArrayPtr bbPtr = forkIO $ do
               threadDelaySecs 30
               grpcDebug "serverRequestCall(R): doing delayed cleanup."

src/Network/GRPC/LowLevel/Op.hs

@@ -20,6 +20,7 @@ import qualified Network.GRPC.Unsafe                   as C (Call)
 import qualified Network.GRPC.Unsafe.ByteBuffer        as C
 import qualified Network.GRPC.Unsafe.Metadata          as C
 import qualified Network.GRPC.Unsafe.Op                as C
+import qualified Network.GRPC.Unsafe.Slice             as C (Slice, freeSlice)
 
 -- | Sum describing all possible send and receive operations that can be batched
 -- and executed by gRPC. Usually these are processed in a handful of
@@ -39,7 +40,7 @@ data Op = OpSendInitialMetadata MetadataMap
 -- 'withOpContexts'.
 data OpContext =
   OpSendInitialMetadataContext C.MetadataKeyValPtr Int
-  | OpSendMessageContext C.ByteBuffer
+  | OpSendMessageContext (C.ByteBuffer, C.Slice)
   | OpSendCloseFromClientContext
   | OpSendStatusFromServerContext C.MetadataKeyValPtr Int C.StatusCode
                                   B.ByteString
@@ -90,7 +91,7 @@ createOpContext OpRecvCloseOnServer =
 setOpArray :: C.OpArray -> Int -> OpContext -> IO ()
 setOpArray arr i (OpSendInitialMetadataContext kvs l) =
   C.opSendInitialMetadata arr i kvs l
-setOpArray arr i (OpSendMessageContext bb) =
+setOpArray arr i (OpSendMessageContext (bb,_)) =
   C.opSendMessage arr i bb
 setOpArray arr i OpSendCloseFromClientContext =
   C.opSendCloseClient arr i
@@ -109,7 +110,8 @@ setOpArray arr i (OpRecvCloseOnServerContext pcancelled) = do
 -- | Cleans up an 'OpContext'.
 freeOpContext :: OpContext -> IO ()
 freeOpContext (OpSendInitialMetadataContext m _) = C.metadataFree m
-freeOpContext (OpSendMessageContext bb) = C.grpcByteBufferDestroy bb
+freeOpContext (OpSendMessageContext (bb, s)) =
+  C.grpcByteBufferDestroy bb >> C.freeSlice s
 freeOpContext OpSendCloseFromClientContext = return ()
 freeOpContext (OpSendStatusFromServerContext metadata _ _ _) =
   C.metadataFree metadata

src/Network/GRPC/Unsafe.chs

@@ -5,6 +5,7 @@ module Network.GRPC.Unsafe where
 import Control.Monad
 
 import Foreign.C.Types
+import Foreign.Marshal.Alloc (free)
 import Foreign.Ptr
 import Foreign.Storable
 
@@ -115,8 +116,15 @@ instance Storable Event where
     {#set grpc_event.success#} p $ if s then 1 else 0
     {#set grpc_event.tag#} p (unTag t)
 
+-- | Used to unwrap structs from pointers. This is all part of a workaround
+-- because the C FFI can't return raw structs directly. So we wrap the C
+-- function, mallocing a temporary pointer. This function peeks the struct
+-- within the pointer, then frees it.
 castPeek :: Storable b => Ptr a -> IO b
-castPeek p = peek (castPtr p)
+castPeek p = do
+  val <- peek (castPtr p)
+  free p
+  return val
 
 {#enum grpc_connectivity_state as ConnectivityState {underscoreToCase}
   deriving (Show, Eq)#}

src/Network/GRPC/Unsafe/ByteBuffer.chs

@@ -84,9 +84,15 @@ withByteStringAsByteBuffer bs f = do
   bracket (byteStringToSlice bs) freeSlice $ \slice -> do
     bracket (grpcRawByteBufferCreate slice 1) grpcByteBufferDestroy f
 
--- TODO: Issue #5
+-- Creates a 'ByteBuffer'. We also return the slice we needed to allocate to
-createByteBuffer :: B.ByteString -> IO ByteBuffer
+-- create it. It is the caller's responsibility to free both when finished using
-createByteBuffer bs = byteStringToSlice bs >>= flip grpcRawByteBufferCreate 1
+-- the byte buffer. In most cases, one should prefer to use
+-- 'withByteStringAsByteBuffer' if possible.
+createByteBuffer :: B.ByteString -> IO (ByteBuffer, Slice)
+createByteBuffer bs = do
+  slice <- byteStringToSlice bs
+  bb <- grpcRawByteBufferCreate slice 1
+  return (bb, slice)
 
 copyByteBufferToByteString :: ByteBuffer -> IO B.ByteString
 copyByteBufferToByteString bb = do

src/Network/GRPC/Unsafe/Slice.chs

@@ -1,3 +1,5 @@
+{-# LANGUAGE StandaloneDeriving #-}
+
 module Network.GRPC.Unsafe.Slice where
 
 #include <grpc/impl/codegen/slice.h>
@@ -12,6 +14,8 @@ import Foreign.Ptr
 -- ByteStrings. This type is a pointer to a C type.
 {#pointer *gpr_slice as Slice newtype #}
 
+deriving instance Show Slice
+
 -- TODO: we could also represent this type as 'Ptr Slice', by doing this:
 -- newtype Slice = Slice {#type gpr_slice#}
 -- This would have no practical effect, but it would communicate intent more

tests/LowLevelTests.hs

@@ -15,6 +15,7 @@ import qualified Network.GRPC.LowLevel.Client.Unregistered as U
 import qualified Network.GRPC.LowLevel.Server.Unregistered as U
 import           Test.Tasty
 import           Test.Tasty.HUnit                          as HU (Assertion,
+                                                                  assertBool,
                                                                   assertEqual,
                                                                   assertFailure,
                                                                   testCase,
@@ -33,6 +34,8 @@ lowLevelTests = testGroup "Unit tests of low-level Haskell library"
   -- , testWrongEndpoint
   , testPayload
   , testPayloadUnregistered
+  , testGoaway
+  , testSlowServer
   ]
 
 testGRPCBracket :: TestTree
@@ -146,12 +149,65 @@ testPayloadUnregistered =
              return ("reply test", mempty, "details string")
       r @?= Right ()
 
+testGoaway :: TestTree
+testGoaway =
+  csTest "Client handles server shutdown gracefully"
+         client
+         server
+         [("/foo", Normal)]
+  where
+    client c = do
+      rm <- clientRegisterMethod c "/foo" Normal
+      clientRequest c rm 10 "" mempty
+      clientRequest c rm 10 "" mempty
+      lastResult <- clientRequest c rm 1 "" mempty
+      assertBool "Client handles server shutdown gracefully" $
+        lastResult == unavailableStatus
+        ||
+        lastResult == Left GRPCIOTimeout
+    server s = do
+      let rm = head (registeredMethods s)
+      serverHandleNormalCall s rm 11 mempty dummyHandler
+      serverHandleNormalCall s rm 11 mempty dummyHandler
+      return ()
+
+testSlowServer :: TestTree
+testSlowServer =
+  csTest "Client handles slow server response" client server [("/foo", Normal)]
+  where
+    client c = do
+      rm <- clientRegisterMethod c "/foo" Normal
+      result <- clientRequest c rm 1 "" mempty
+      assertBool "Client gets timeout or deadline exceeded" $
+        result == Left GRPCIOTimeout
+        ||
+        result == deadlineExceededStatus
+    server s = do
+      let rm = head (registeredMethods s)
+      serverHandleNormalCall s rm 1 mempty $ \_ _ -> do
+        threadDelay (2*10^(6 :: Int))
+        return ("", mempty, mempty, StatusDetails "")
+      return ()
+
 --------------------------------------------------------------------------------
 -- Utilities and helpers
 
 dummyMeta :: M.Map ByteString ByteString
 dummyMeta = [("foo","bar")]
 
+dummyHandler :: ByteString -> MetadataMap
+                -> IO (ByteString, MetadataMap, MetadataMap, StatusDetails)
+dummyHandler _ _ = return ("", mempty, mempty, StatusDetails "")
+
+unavailableStatus :: Either GRPCIOError a
+unavailableStatus =
+  Left (GRPCIOBadStatusCode GrpcStatusUnavailable (StatusDetails ""))
+
+deadlineExceededStatus :: Either GRPCIOError a
+deadlineExceededStatus =
+  Left (GRPCIOBadStatusCode GrpcStatusDeadlineExceeded
+                            (StatusDetails "Deadline Exceeded"))
+
 nop :: Monad m => a -> m ()
 nop = const (return ())
 

tests/UnsafeTests.hs

@@ -142,8 +142,11 @@ payloadClient = do
             destroyStatusCodePtr statusCodePtr
         --verify response received
         responseRecv <- peek clientRecvBB
-        responseRecvBS <- copyByteBufferToByteString responseRecv
+        let (ByteBuffer rawPtr) = responseRecv
-        responseRecvBS HU.@?= "hello you"
+        if rawPtr == nullPtr
+           then error "Client got null pointer for received response!"
+           else do responseRecvBS <- copyByteBufferToByteString responseRecv
+                   responseRecvBS HU.@?= "hello you"
         grpcCompletionQueueShutdown cq
         grpcCallDestroy clientCall
         --TODO: the grpc test drains the cq here