-- | Unlike most of the other internal low-level modules, we don't export -- everything here. There are several things in here that, if accessed, could -- cause race conditions, so we only expose functions that are thread safe. -- However, some of the functions we export here can cause memory leaks if used -- improperly. -- -- When definition operations which pertain to calls, this module only provides -- definitions for registered calls; for unregistered variants, see -- `Network.GRPC.LowLevel.CompletionQueue.Unregistered`. Type definitions and -- implementation details to both are kept in -- `Network.GRPC.LowLevel.CompletionQueue.Internal`. {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiWayIf #-} {-# LANGUAGE PolyKinds #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE ViewPatterns #-} module Network.GRPC.LowLevel.CompletionQueue ( CompletionQueue , withCompletionQueue , createCompletionQueue , shutdownCompletionQueue , pluck , startBatch , channelCreateCall , TimeoutSeconds , isEventSuccessful , serverRegisterCompletionQueue , serverShutdownAndNotify , serverRequestCall , newTag ) where import Control.Concurrent.STM.TVar (newTVarIO) import Control.Exception (bracket) import Control.Monad.Managed import Control.Monad.Trans.Class (MonadTrans (lift)) import Control.Monad.Trans.Except import Data.IORef (newIORef) import Data.List (intersperse) import Foreign.Ptr (nullPtr) import Foreign.Storable (peek) import Network.GRPC.LowLevel.Call import Network.GRPC.LowLevel.CompletionQueue.Internal import Network.GRPC.LowLevel.GRPC import qualified Network.GRPC.Unsafe as C import qualified Network.GRPC.Unsafe.Constants as C import qualified Network.GRPC.Unsafe.Metadata as C import qualified Network.GRPC.Unsafe.Op as C import qualified Network.GRPC.Unsafe.Time as C withCompletionQueue :: GRPC -> (CompletionQueue -> IO a) -> IO a withCompletionQueue grpc = bracket (createCompletionQueue grpc) shutdownCompletionQueue createCompletionQueue :: GRPC -> IO CompletionQueue createCompletionQueue _ = do unsafeCQ <- C.grpcCompletionQueueCreateForPluck C.reserved currentPluckers <- newTVarIO 0 currentPushers <- newTVarIO 0 shuttingDown <- newTVarIO False nextTag <- newIORef minBound return CompletionQueue{..} -- | Very simple wrapper around 'grpcCallStartBatch'. Throws 'GRPCIOShutdown' -- without calling 'grpcCallStartBatch' if the queue is shutting down. -- Throws 'CallError' if 'grpcCallStartBatch' returns a non-OK code. startBatch :: CompletionQueue -> C.Call -> C.OpArray -> Int -> C.Tag -> IO (Either GRPCIOError ()) startBatch cq@CompletionQueue{..} call opArray opArraySize tag = withPermission Push cq $ fmap throwIfCallError $ do grpcDebug $ "startBatch: calling grpc_call_start_batch with pointers: " ++ show call ++ " " ++ show opArray res <- C.grpcCallStartBatch call opArray opArraySize tag C.reserved grpcDebug "startBatch: grpc_call_start_batch call returned." return res channelCreateCall :: C.Channel -> Maybe (ServerCall a) -> C.PropagationMask -> CompletionQueue -> C.CallHandle -> C.CTimeSpecPtr -> IO (Either GRPCIOError ClientCall) channelCreateCall chan parent mask cq@CompletionQueue{..} handle deadline = withPermission Push cq $ do let parentPtr = maybe (C.Call nullPtr) unsafeSC parent grpcDebug $ "channelCreateCall: call with " ++ concat (intersperse " " [show chan, show parentPtr, show mask, show unsafeCQ, show handle, show deadline]) call <- C.grpcChannelCreateRegisteredCall chan parentPtr mask unsafeCQ handle deadline C.reserved return $ Right $ ClientCall call -- | Create the call object to handle a registered call. serverRequestCall :: RegisteredMethod mt -> C.Server -> CompletionQueue -- ^ server CQ -> CompletionQueue -- ^ call CQ -> IO (Either GRPCIOError (ServerCall (MethodPayload mt))) serverRequestCall rm s scq ccq = -- NB: The method type dictates whether or not a payload is present, according -- to the payloadHandling function. We do not allocate a buffer for the -- payload when it is not present. withPermission Push scq . with allocs $ \(dead, call, pay, meta) -> withPermission Pluck scq $ do md <- peek meta tag <- newTag scq dbug $ "got pluck permission, registering call for tag=" ++ show tag ce <- C.grpcServerRequestRegisteredCall s (methodHandle rm) call dead md pay (unsafeCQ ccq) (unsafeCQ scq) tag runExceptT $ case ce of C.CallOk -> do ExceptT $ do r <- pluck' scq tag Nothing dbug $ "pluck' finished:" ++ show r return r lift $ ServerCall <$> peek call <*> return ccq <*> C.getAllMetadataArray md <*> extractPayload rm pay <*> convertDeadline dead _ -> do lift $ dbug $ "Throwing callError: " ++ show ce throwE (GRPCIOCallError ce) where allocs = (,,,) <$> mgdPtr <*> mgdPtr <*> mgdPayload (methodType rm) <*> managed C.withMetadataArrayPtr dbug = grpcDebug . ("serverRequestCall(R): " ++) convertDeadline timeSpecPtr = C.timeSpec <$> peek timeSpecPtr -- | Register the server's completion queue. Must be done before the server is -- started. serverRegisterCompletionQueue :: C.Server -> CompletionQueue -> IO () serverRegisterCompletionQueue server CompletionQueue{..} = C.grpcServerRegisterCompletionQueue server unsafeCQ C.reserved serverShutdownAndNotify :: C.Server -> CompletionQueue -> C.Tag -> IO () serverShutdownAndNotify server CompletionQueue{..} tag = C.grpcServerShutdownAndNotify server unsafeCQ tag